Async Database¶

The fluid.db module provides a simple asynchronous interface to interact with postgres databases. It is built on top of the sqlalchemy and asyncpg libraries.

Installation¶

To use the database module, you need to install the db extra, and optionally the cli extra if you want to use the command line interface for managing database migrations:

pip install aio-fluid[db,cli]

Database¶

There are two database classes:

Database — provides connection management, transactions, and migrations.
CrudDB — extends Database with CRUD helpers for common query patterns.

Most applications should use CrudDB directly:

from fluid.db import CrudDB

db = CrudDB("postgresql+asyncpg://postgres:postgres@localhost:5432/mydb")

Note the use of the postgresql+asyncpg driver in the connection string. This is required for the async engine.

You can also load the DSN from an environment variable (defaults to DATABASE):

db = CrudDB.from_env()

Connections are pooled. The pool size is set by the FLUID_DBPOOL_MAX_SIZE and FLUID_DBPOOL_MAX_OVERFLOW environment variables, and each pooled connection is tested when it is taken from the pool (FLUID_DBPOOL_PRE_PING, on by default), so connections the server has dropped, for instance on a database restart, are replaced rather than failing the next query. See settings for how settings are read from the environment.

Register a Table¶

Register tables against the database's metadata so that migrations and CRUD helpers can discover them:

import sqlalchemy as sa

articles = sa.Table(
    "articles",
    db.metadata,
    sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
    sa.Column("title", sa.String(200), nullable=False),
    sa.Column("author", sa.String(100)),
    sa.Column("score", sa.Integer),
    sa.Column("published_at", sa.DateTime(timezone=True)),
)

Migrations¶

The Migration object wraps Alembic and is obtained from the database:

mig = db.migration()

Create the database¶

# creates the database if it doesn't exist yet; returns False if it already existed
mig.db_create()

Apply migrations¶

# upgrade to the latest revision (equivalent to `alembic upgrade heads`)
mig.upgrade("heads")

For quick setup in tests or development, you can create all tables directly from metadata without Alembic revision files:

mig.create_all()

Generate a new revision¶

# auto-generate a revision by diffing metadata against the current schema
mig.revision("add score column", autogenerate=True)

Connections and Transactions¶

Use connection() when you only need to read data:

async with db.connection() as conn:
    result = await conn.execute(sa.text("SELECT 1"))

Use transaction() when you need to write data — it commits on exit and rolls back on exception:

async with db.transaction() as conn:
    await conn.execute(articles.insert().values(title="Hello"))

The ensure_connection and ensure_transaction variants are useful in functions that may receive an existing connection from the caller, avoiding nested connections:

async def insert_article(data: dict, conn=None):
    async with db.ensure_transaction(conn) as conn:
        await conn.execute(articles.insert().values(**data))

CRUD Operations¶

CrudDB provides async helpers that cover the most common patterns.

Insert¶

# single row
row = (await db.db_insert(articles, {"title": "Hello", "score": 10})).one()

# multiple rows — missing columns are filled with None automatically
rows = (await db.db_insert(articles, [
    {"title": "Hello", "score": 10},
    {"title": "World"},
])).fetchall()

All insert operations return the inserted rows via RETURNING *.

Select¶

rows = (await db.db_select(articles, {"author": "alice"})).fetchall()

Pass order_by to sort results. Prefix a field name with - for descending order:

rows = (await db.db_select(articles, {}, order_by=("-published_at",))).fetchall()

Pass columns to select only some columns, rather than every column of the table. This avoids reading large columns you do not need, and accepts computed expressions too:

from sqlalchemy import func

rows = (
    await db.db_select(
        articles,
        {"author": "alice"},
        columns=[articles.c.id, func.lower(articles.c.title).label("title")],
    )
).fetchall()

Filter operators¶

Filters use a "field:op" key syntax. The default operator is eq:

Key	Meaning
`"score"` or `"score:eq"`	`score = value`
`"score:ne"`	`score != value`
`"score:gt"`	`score > value`
`"score:ge"`	`score >= value`
`"score:lt"`	`score < value`
`"score:le"`	`score <= value`

Pass a list as the value to match any of its values with eq, or none of them with ne:

# WHERE score = ANY($1), with $1 the array [5, 10, 15]
rows = (await db.db_select(articles, {"score": [5, 10, 15]})).fetchall()

The list is sent as a single array parameter, = ANY for eq and <> ALL for ne, rather than one parameter per value, so it can hold any number of values: a query can carry at most 32767 parameters. An empty list matches no row with eq and excludes none with ne.

Update¶

rows = (await db.db_update(articles, {"author": "alice"}, {"score": 99})).fetchall()

Returns all updated rows via RETURNING *.

Upsert¶

db_upsert updates a single record if it exists, or inserts it if it does not:

# update score if the row exists, otherwise insert it
row = await db.db_upsert(
    articles,
    {"title": "Hello"},   # lookup key
    {"score": 42},        # data to set
)

The upsert is atomic: it runs a single INSERT ... ON CONFLICT (...) DO UPDATE statement, so concurrent upserts of the same record never clash. Two rules follow from that statement:

the lookup key columns must match a primary key or unique constraint of the table (here title must be unique), and filter operators such as score:gt are not supported;
the lookup key and data together must be a valid row to insert, since Postgres checks NOT NULL constraints before it detects the conflict.

To upsert several records at once, use db_upsert_many with full rows and the key columns:

rows = await db.db_upsert_many(
    articles,
    [
        {"title": "Hello", "score": 42},
        {"title": "World", "score": 7},
    ],
    key=("title",),
)

The same rules apply, plus a few more:

every record must have the same columns, so that an update never overwrites a column with NULL because a record left it out;
no two records can share a key, since Postgres cannot update the same row twice in one statement;
the returned rows do not follow the order of the records, so match them by key.

Large lists are sent in batches of batch_size records (1000 by default), all within one transaction, so either every record is upserted or none is.

Delete¶

deleted = (await db.db_delete(articles, {"author": "alice"})).fetchall()

Returns all deleted rows via RETURNING *.

Count¶

n = await db.db_count(articles, {"author": "alice"})

Pagination¶

Pagination implements cursor-based pagination on top of CrudDB. It fetches one extra row beyond the requested limit to determine whether a next page exists, then encodes a cursor that the client returns with the next request.

from fluid.db import Pagination

# first page
rows, cursor = await Pagination.create(
    "published_at",
    "id",
    limit=20,
    filters={"author": "alice"},
    desc=True,
).execute(db, articles)

# next page, the limit and the filters of the first page are in the cursor
if cursor:
    rows, cursor = await Pagination.create(
        "published_at",
        "id",
        cursor=cursor,
        filters={"author": "alice"},
        desc=True,
    ).execute(db, articles)

When cursor is provided, the limit argument is ignored: the limit is decoded from the cursor, along with the filters and the search text of the first page, so the pages stay consistent when a client sends the cursor alone.

Signing the cursor¶

A cursor is base64 encoded JSON, so a client can decode it, change it and send it back. Set FLUID_PAGINATION_SECRET (the pagination_secret field of Settings) and every cursor is signed with an HMAC of its content: a cursor that is changed, unsigned or signed with another secret is rejected with a ValidationError. Every process serving the same clients needs the same secret, otherwise a cursor issued by one is rejected by another. Changing the secret invalidates the cursors already issued, and the clients start again from the first page.

Signed or not, two rules keep a cursor from reaching rows the caller must not see:

The filters passed to create are applied on top of the filters in the cursor, and win over them on the same key. Pass every filter the caller enforces, the owner of the rows or a public flag for instance, on every request, not only on the first page.
A cursor limit below 1 or above DEFAULT_PAGINATION_MAX_LIMIT (500 by default) is replaced by that maximum, so a cursor cannot ask for an unbounded page.

Without a secret these rules are the only protection: a filter that is only in the cursor can be removed by the client.

The ordering must be unique¶

The ordering fields must uniquely identify a row. The cursor encodes only the values of those fields for the first row of the next page, and the next query resumes from that position, so ties are ambiguous: rows sharing the same ordering values can appear on two consecutive pages, and rows can be skipped altogether, because the database is free to order tied rows differently between the two queries.

published_at alone is not unique — two articles can be published at the same instant — which is why the examples above order by ("published_at", "id"). Appending the primary key makes the ordering total and the cursor unambiguous. The same rule applies to the ordering used with full-text search below.

Full-text search¶

Combine pagination with a full-text search across multiple columns:

from fluid.db import Pagination
from fluid.db.pagination import Search

rows, cursor = await Pagination.create(
    "published_at",
    "id",
    limit=20,
    search=Search(search_fields=("title", "author"), search_text="fluid"),
    desc=True,
).execute(db, articles)

Selecting columns¶

By default every column of the table is selected. Pass columns to execute to select only some of them, for instance to leave out large columns a listing does not show. The cursor is built from the ordering fields of the last row, so the columns must include them:

rows, cursor = await Pagination.create(
    "published_at",
    "id",
    limit=20,
).execute(
    db,
    articles,
    columns=[articles.c.id, articles.c.published_at, articles.c.title],
)