Skip to content

Database & Migrations

Defining a model

# app/Models/post.py
from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column

from zeython import Model

class Post(Model):
    __tablename__ = "posts"

    title: Mapped[str] = mapped_column(String(255))
    body: Mapped[str] = mapped_column(Text)

Every Model subclass already has id, created_at, updated_at, is_deleted, and deleted_at columns — you only declare the columns specific to your table.

Register new models in app/Models/__init__.py (done automatically by zeython make model) so Alembic's autogenerate can see them.

Active Record API

post = await Post.create(title="Hello", body="World")
post = await Post.find(1)
posts = await Post.all()
posts = await Post.find_by(title="Hello")
await post.update(title="Updated")
await post.delete()                 # soft delete by default
await post.delete(soft=False)       # hard delete
await post.restore()
post.to_dict()                      # JSON-serializable dict

All of these require an active database session — present automatically inside a request, or via async with database.session(): elsewhere (see Architecture).

find/all/find_by/first_where also accept include=("relationship_name",) to eager-load relationships — required reading before you define your first relationship(), since touching one without eager-loading crashes async code differently than you'd expect from sync SQLAlchemy. See Relationships.

A model that declares a tenant_id column gets every one of these methods scoped to the current tenant automatically — see Multi-Tenancy.

Mass-assignment protection

create()/update() accept **kwargs, which makes it tempting to spread a request body straight into them:

data = await request.json()
post = await Post.create(**data)   # every key in data becomes a column

That's fine for a Post with only title/body — but the moment a model gains a column that shouldn't be settable from a request (is_admin, role, balance, approved, ...), the exact same pattern lets a caller set that too, just by including it in the JSON body. This is the classic mass-assignment vulnerability.

Two independent protections apply automatically, only to the **kwargs path — direct attribute assignment (user.is_admin = True) is never affected:

  • id, created_at, updated_at, is_deleted, and deleted_at are always guarded, on every model — Post.create(title="x", id=999) silently ignores id.
  • List any of your own model's sensitive columns in __guarded__, and create()/update() silently drop those too:
class User(Model):
    __tablename__ = "users"
    __guarded__ = ("is_admin",)

    name: Mapped[str] = mapped_column(String(255))
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)

user = await User.create(**data)   # data = {"name": "Eve", "is_admin": True}
user.is_admin                      # False -- silently dropped, not set

To actually grant admin, set it directly from trusted code instead of through the mass-assignment path: user.is_admin = True; await user.save(), or await user.update(name="Eve") followed by that same direct assignment.

__guarded__ is opt-in and empty by default — declare it for any column you'd never want set from a raw, unfiltered **data spread. The safer alternative that needs no declaration at all is to keep pulling specific fields out of the request body by name (Post.create(title=data.get("title"), body=data.get("body"))), which is what every controller example in this documentation other than this one does.

__guarded__ only wraps create()/update() -- constructing a model directly (User(**data)) bypasses it entirely, going straight to SQLAlchemy's own constructor. Prefer create() over direct construction for exactly this reason; if you do construct directly (as Authentication's registration example does, since it needs to call set_password() before the first save()), pull fields out by name there too.

Transactions

Every request already runs inside one implicit transaction: DatabaseSessionMiddleware opens a session at the start of the request and commits it if the response status ends up below 400, or rolls it back otherwise -- whether that's a genuinely unhandled exception unwinding past the middleware, or one of zeython's own HTTPException subclasses (NotFoundException, ValidationException, etc.) that your handler raised and Starlette's inner ExceptionMiddleware turned into a response without ever re-raising past this middleware. Only the response status reveals that case, which is exactly why the middleware checks it instead of relying on exception propagation alone. Nothing extra is needed for "undo everything this request did if it fails":

async def transfer(self, request):
    await from_account.update(balance=from_account.balance - amount)
    await to_account.update(balance=to_account.balance + amount)
    if something_goes_wrong:
        raise ConflictException("Transfer failed")
        # both updates above are rolled back -- the whole request's
        # writes are, whenever an exception ends it

transaction() is for a narrower case: isolating part of a request so a failure there doesn't undo everything else, without ending the request:

from zeython import transaction

async def checkout(self, request):
    order = await Order.create(user_id=user.id, status="pending")

    try:
        async with transaction():
            await reserve_inventory(order)   # several writes
            await charge_payment(order)      # might raise
    except PaymentFailedException:
        await order.update(status="payment_failed")
        return JSONResponse({"error": "Payment failed"}, status_code=402)

    await order.update(status="confirmed")
    return JSONResponse(order.to_dict())

If reserve_inventory/charge_payment raise, only their writes roll back (a SAVEPOINT under the hood) -- order's initial creation isn't touched, and the handler keeps running to record the failure and respond normally, rather than the whole request dying with a 500. transaction() blocks nest: an inner one rolling back doesn't affect an outer one still in progress.

Requires an active session, same as the rest of the Active Record API -- raises the same RuntimeError as calling Model.create() outside one.

Locking a row

find(id, for_update=True) locks the row for the rest of the current transaction (SELECT ... FOR UPDATE) -- a second, concurrent find(..., for_update=True) for the same row elsewhere blocks until this transaction commits or rolls back, instead of both transactions reading the same not-yet-committed state and independently making the same decision from it:

async def redeem(self, request):
    coupon = await Coupon.find(coupon_id, for_update=True)
    if coupon.redeemed:
        raise ConflictException("Already redeemed.")
    await coupon.update(redeemed=True)

Without the lock, two requests racing to redeem the same coupon could both read redeemed=False before either had written True, and both would succeed -- zeython.mfa.verify_and_consume() uses this exact pattern to stop two requests from spending the same one-time recovery code. Reach for this whenever a decision depends on a row's current state and getting it wrong twice is a real problem (redeeming something one-time-use, decrementing a limited stock count) -- not for every read just in case, since a lock held across the rest of a slow request blocks whoever's waiting on it for that whole time.

SQLite has no row-level locking -- for_update=True there compiles away to an ordinary, unlocked SELECT (and logs a warning saying so): two concurrent SQLite readers still both see the pre-write state, same as without the flag. Postgres and MySQL support it fully.

Pagination

all() loads every matching row — fine for a small table, not for a listing endpoint whose table grows without bound. paginate() is the same query, sliced:

page = await Post.paginate(page=1, per_page=20)

page.items         # list[Post] -- this page's rows
page.page          # 1
page.per_page      # 20
page.total         # every matching row, not just this page
page.total_pages   # ceil(total / per_page)
page.has_next       # page < total_pages
page.has_prev       # page > 1

total costs a second query (a COUNT(*) over the same filters) to compute — that's the price of knowing total_pages/has_next up front, not a bug. If you don't need that, all() with a hand-rolled limit isn't available on Model directly, but nothing stops you from writing a raw select() for that one case.

paginate() accepts the same include_deleted/include=(...) keywords as find/all/find_by. zeython new wires it into the generated GET /users (?page=/?per_page=, defaulting to 1/20) — see app/Controllers/user_controller.py.

page.to_dict() serializes a whole page in one call — items (via each item's own to_dict(), for Model instances) plus the metadata above:

return JSONResponse(page.to_dict())
# {"items": [...], "page": 1, "per_page": 20, "total": 57,
#  "total_pages": 3, "has_next": true, "has_prev": false}

Pass the current request to also get next_url/prev_url — the same URL with only the page query param changed (every other query param carries over), None when there is no next/previous page:

return JSONResponse(page.to_dict(request=request))
# adds "next_url": "http://.../users?page=2", "prev_url": null

Connection pooling

DatabaseServiceProvider forwards DATABASE_POOL_SIZE/DATABASE_MAX_OVERFLOW straight through to SQLAlchemy's connection pool, unset by default:

DATABASE_POOL_SIZE=10
DATABASE_MAX_OVERFLOW=20
  • DATABASE_POOL_SIZE — steady-state connections the pool keeps open.
  • DATABASE_MAX_OVERFLOW — extra connections allowed beyond that under load, closed again once things quiet down.

Meaningful for PostgreSQL/MySQL, and for a file-based SQLite URL (what zeython new scaffolds, sqlite+aiosqlite:///./database.db) — all three default to SQLAlchemy's AsyncAdaptedQueuePool, which both settings configure directly. The one exception is sqlite+aiosqlite:///:memory: (what the framework's own test suite uses): in-memory SQLite defaults to StaticPool, which doesn't accept either kwarg at all — passing them raises TypeError at engine-creation time. That's why both are unset by default rather than shipping a number that would break an in-memory setup; set them once you have an actual concurrency figure to size against (a reasonable start: your app server's worker count, or a little above it).

Read replicas

A second DATABASE_READ_URL routes read-heavy work to a replica instead of the primary — a report, a dashboard, an analytics query, anything that can tolerate a little replication lag and that you'd rather not have competing with write traffic for the primary's connections:

DATABASE_URL=postgresql+asyncpg://user:pass@primary/app
DATABASE_READ_URL=postgresql+asyncpg://user:pass@replica/app

async def monthly_report(self, request):
    async with database.read_replica():
        orders = await Order.all()
    return JSONResponse(build_report(orders))

read_replica() opens a session against the replica exactly the way database.session() opens one against the primary — same current_session() underneath, so Model.find/all/find_by/... all work inside the block unchanged. Not registered as the request's default session — reach for it explicitly, only for the read path you actually want off the primary; everything else in the request still uses the regular session.

Read-only in practice, not by any check this framework adds. A real replica is normally configured read-only at the database level itself (Postgres's default_transaction_read_only, a MySQL replica user with no write grants) — a write attempted inside read_replica() fails with a real database error there, the same as it would against any other client connected to that replica. There's also no commit() on exit, since a replica session exists for reads.

Optional — no DATABASE_READ_URL set, read_replica() transparently falls back to opening a session against the primary. Code written against it works the same whether or not a replica is actually configured, so it's safe to write async with database.read_replica(): around a report query in an app that doesn't have one yet.

Passed through via **engine_kwargs on Database.__init__ — the same mechanism accepts any other keyword create_async_engine() understands, if you construct Database yourself instead of going through DatabaseServiceProvider.

Migrations

zeython db revision -m "add posts table"
zeython db migrate

zeython new scaffolds a working Alembic setup (alembic.ini, migrations/env.py) pointed at your DATABASE_URL and your app.Models metadata, so --autogenerate works out of the box against SQLite, PostgreSQL (pip install zeython[postgres]), or MySQL (pip install zeython[mysql]). migrations/env.py also enables Alembic's render_as_batch mode, needed for SQLite specifically: SQLite can add a new column but can't otherwise ALTER a constraint in place, so a migration adding a ForeignKey (or any other constraint change) to an existing table would fail without it. Harmless no-op on Postgres/MySQL.

Adding a NOT NULL column to a table that already has rows hits a real SQL constraint on every database, not a Zeython limitation: the existing rows need some value for that column. Give the generated migration a default — sa.Column('author_id', sa.Integer(), nullable=False, server_default='1') — so the backfill has something to write, or make the column nullable if the data genuinely doesn't apply to old rows.