Migrations¶

Every database recipe assumes the schema already exists. This one is about the step before that: how the schema comes into being, and why the hand-written version of that step is wrong in a way that stays invisible for weeks.

The SDK ships the whole path as one method — AlembicHelper.sync_schema() — and the rest of this page explains what it decides, so you can recognise which state your database is in.

The complete bootstrap¶

# src/db/schema.py
from tempest_fastapi_sdk import AlembicHelper, SchemaSyncOutcome

from src.core.settings import settings


async def sync_schema() -> SchemaSyncOutcome:
    """Bring the database schema in line with the migration tree."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    return await helper.sync_schema_async()

Call it from the lifespan and the service boots with the right schema from any starting state:

# src/api/app.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from src.db.schema import sync_schema


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Sync the schema before the first request is served."""
    outcome = await sync_schema()
    print(f"schema: {outcome.value}")
    yield


app: FastAPI = FastAPI(lifespan=lifespan)

From async code, use the `_async` method¶

The lifespan runs inside uvicorn's event loop. Every AlembicHelper method that executes alembic/env.py has an _async twin — upgrade_async, safe_upgrade_async, downgrade_async, stamp_async, revision_async, check_async, current_async, has_existing_schema_async, adopt_async, sync_schema_async, squash_async, pending_destructive_ops_async — with the same signature, and that is the one you call from async code.

The reason is the env.py the SDK generates: it starts the async engine with asyncio.run(...), and asyncio.run cannot be called while a loop is already running. Calling the sync method from the lifespan now fails immediately, with the right instruction:

RuntimeError: AlembicHelper.upgrade() was called from a running event loop. It runs alembic/env.py, which drives migrations with asyncio.run() and cannot nest inside the loop; use `await helper.upgrade_async(...)` instead.

Before, the same error came from deep inside Alembic as asyncio.run() cannot be called from a running event loop, together with a RuntimeWarning: coroutine 'run_async_migrations' was never awaited — and check() did not even raise: it swallowed the error and answered False, as if the schema had drifted.

The _async method runs the sync one in a worker thread, which has no loop of its own, so the asyncio.run in env.py works there and the service's loop keeps serving while the migration runs. Since the thread needs nothing new in env.py, the env.py already in your repository keeps working — no need to regenerate it.

Sync current() still works on the loop

Reading the revision does not go through env.py: with a sync driver installed (the stdlib sqlite3, or psycopg2 on PostgreSQL), helper.current() works from async code as before. Only an async-only install (asyncpg with no sync driver) takes the asyncio.run path — and then it raises too, asking for current_async().

Technical details: sharing the connection instead of a thread

Alembic has an official recipe for running the migration on the loop itself: open an AsyncConnection and hand the sync connection that run_sync provides over as config.attributes["connection"]. The env.py the SDK generates from this release on accepts it — given the connection, it migrates on it, without creating an engine or calling asyncio.run:

# src/db/shared_connection.py
from alembic import command
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from tempest_fastapi_sdk import AlembicHelper

from src.core.settings import settings


async def upgrade_on_shared_connection() -> None:
    """Run every pending migration on a connection this code owns."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    config = helper.config

    def _upgrade(connection: Connection) -> None:
        """Hand the sync connection to env.py and upgrade on it."""
        config.attributes["connection"] = connection
        command.upgrade(config, "head")

    engine: AsyncEngine = create_async_engine(settings.DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(_upgrade)
    await engine.dispose()

The _async methods do not take that path, and the reason is compatibility: an env.py generated before this release ignores config.attributes and calls asyncio.run regardless — inside run_sync the loop is running, and it fails with the same error as before. The thread works with every env.py already generated. Use the path above when the migration must run on your connection or transaction, and regenerate env.py for it: run tempest db init in an empty directory and copy the generated alembic/env.py over yours, checking the metadata import (the default is from src.db.models import BaseModel).

The new env.py also refuses, with its own message, to be executed from inside a loop without a connection handed over — the case of calling command.upgrade straight from async code.

The three states it tells apart¶

sync_schema() asks the question almost every hand-written bootstrap forgets: does the database hold tables Alembic did not create?

Starting state	What runs	Return
Empty database	`safe_upgrade()` — the base revision creates the tables	`SchemaSyncOutcome.SYNCED`
Database that predates Alembic	stamps the base, then `safe_upgrade()`	`SchemaSyncOutcome.ADOPTED`
Database already under Alembic	`safe_upgrade()`	`SchemaSyncOutcome.SYNCED`
Project with no revisions yet	nothing	`SchemaSyncOutcome.NO_MIGRATIONS`

The empty database is the easy case, and it is what makes the rest work: the schema the base revision builds is, by construction, the same one later revisions assume they are altering — because it came from them.

Why `create_tables()` is not the answer¶

The Database recipe says db.create_tables() is for tests and local development only. It is worth also saying what happens if you use it, because the prohibition without the consequence does not make anyone see the defect in their own code:

create_all is CREATE TABLE IF NOT EXISTS

Against a table that already exists, create_tables() adds no column at all. It does not fail, does not warn, and does not return anything different. It is a silent no-op.

The real defect, which took a service down for a day:

# scripts/broken_bootstrap.py — the defect, reproduced; not a recipe.
from tempest_fastapi_sdk import AlembicHelper, AsyncDatabaseManager

from src.core.settings import settings

db: AsyncDatabaseManager = AsyncDatabaseManager(settings.DATABASE_URL)


async def broken_bootstrap() -> None:
    """Reach the worst possible state: old schema, Alembic reporting head."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    if await helper.current_async() is None:
        await db.create_tables()
        await helper.stamp_async("head")
        return
    await helper.safe_upgrade_async()

Every line is plausible. Together they produce the worst possible state: an old schema, and Alembic declaring itself up to date.

$ alembic current
a3f9c21e88b4 (head)

$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

Nothing to do. alembic history flags nothing. Weeks later, the first query touching a new column blows up far from the cause:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such column: messages.edited_at

On the user's side that reads as "sending messages stopped working": the history GET returns 500, and nothing at boot said a word.

Adopting a database that predates Alembic¶

This is the case the code above was trying to handle. The right answer is not to stamp head — it is to stamp the base revision:

# src/db/schema.py
from tempest_fastapi_sdk import AlembicHelper

from src.core.settings import settings


def adopt_existing_database() -> bool:
    """Bring a pre-Alembic schema under Alembic, without upgrading it."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    return helper.adopt()

Stamping the base says "the baseline is already applied" — which is true, because the tables exist — and leaves every revision after it pending, which is also true. The safe_upgrade() that follows runs exactly those.

Stamping head says "everything is already applied", which is false for all but the first.

adopt() does nothing when that is not the case: a database already stamped has nothing to adopt, and an empty one needs the baseline to run, not to be skipped. That is why sync_schema() can call it unconditionally.

The question is answered by has_existing_schema()

It lists the tables and discounts alembic_version, which Alembic writes itself — its presence says nothing about the application schema.

Repairing a database stamped wrong¶

If you are already in the bad state, the fix has two steps. The first clears the pointer; the second re-adopts properly:

# scripts/repair_schema.py
from tempest_fastapi_sdk import AlembicHelper, SchemaSyncOutcome

from src.core.settings import settings


def repair() -> SchemaSyncOutcome:
    """Clear a wrong stamp and re-adopt the schema from the base."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    helper.stamp("base")
    return helper.sync_schema()

stamp("base") is not the same as stamp(helper.base_revision()): "base" is Alembic's word for no revision applied, and it deletes the alembic_version row. After that the database looks like what it is — an existing schema with no pointer — and sync_schema() takes the adoption path.

Check what is pending before you upgrade

After stamp("base"), the revisions between base and head will run against a schema that may already carry part of them. Run helper.pending_destructive_ops() and read helper.history() first — and have a backup. safe_upgrade refuses destructive migrations without force=True, which helps, but does not replace looking.

SQLite: migrations run with foreign keys off¶

On SQLite, Alembic alters a table in batch mode (the generated env.py sets render_as_batch=True): it creates a copy with the new shape, copies the rows, drops the original and renames the copy. With PRAGMA foreign_keys=ON, that DROP TABLE runs the foreign-key actions of the dropped table — and a child with ON DELETE CASCADE goes with it.

Measured with Alembic 1.19.1 and SQLite 3.47.1, adding one nullable column to the parent table with batch_alter_table("org", recreate="always"):

`PRAGMA foreign_keys`	Rows in `member` before → after	Error
`0`	3 → 3	none
`1`	3 → 0	none

So the engines the SDK builds to migrate — the generated env.py's and the AlembicHelper read engines (current(), has_existing_schema()) — keep FK off, while AsyncDatabaseManager turns it on by default. If you use the connection-sharing recipe (config.attributes["connection"], described above), the generated env.py checks the pragma of the connection it receives and refuses before migrating:

RuntimeError: refusing to run migrations on a SQLite connection with PRAGMA foreign_keys=ON: Alembic batch mode recreates tables, and dropping the old copy of a parent table runs ON DELETE CASCADE on its children, deleting their rows without an error. ...

The way out is migrating on a connection from an engine without FK — a manager just for that:

# src/db/shared_connection.py
from alembic import command
from sqlalchemy.engine import Connection
from tempest_fastapi_sdk import AlembicHelper, AsyncDatabaseManager

from src.core.settings import settings


async def upgrade_on_shared_connection() -> None:
    """Migrate on a connection that does not enforce foreign keys."""
    helper: AlembicHelper = AlembicHelper(
        "alembic.ini",
        db_url=settings.DATABASE_URL,
    )
    config = helper.config

    def _upgrade(connection: Connection) -> None:
        """Hand the connection to env.py and migrate on it."""
        config.attributes["connection"] = connection
        command.upgrade(config, "head")

    migrations_db: AsyncDatabaseManager = AsyncDatabaseManager(
        settings.DATABASE_URL,
        sqlite_foreign_keys=False,
    )
    await migrations_db.connect()
    try:
        async with migrations_db.engine.begin() as conn:
            await conn.run_sync(_upgrade)
    finally:
        await migrations_db.disconnect()

PRAGMA foreign_keys=OFF cannot fix an open connection

SQLite ignores the pragma inside a transaction, and the connection that reaches env.py is already in one. Turning it off has to happen when the engine is built (sqlite_foreign_keys=False, or a plain create_async_engine(url)), not on the connection.

An env.py generated before this version does not refuse

The check lives in env.py, and yours was copied into the repository when you ran tempest db init. If it receives a connection from an AsyncDatabaseManager — which now turns FK on by default —, switch to a connection without FK as above and regenerate env.py (run tempest db init in an empty directory and copy alembic/env.py over yours, checking the metadata import). Migrating through AlembicHelper or the CLI is not affected: those paths open their own engine, without FK.

The generated `env.py`¶

tempest db init (or AlembicHelper.init) writes an alembic/env.py of three statements — the docstring is the only prose, with no inline comment:

"""Alembic environment script generated by tempest-fastapi-sdk."""

from tempest_fastapi_sdk.db import run_alembic_env

from src.db.models import BaseModel

target_metadata = BaseModel.metadata

run_alembic_env(target_metadata)

Everything the file used to do now lives in run_alembic_env: resolving DATABASE_URL, the revision hooks, render_enum_types, batch mode, the offline and online paths, and the connection handed over in config.attributes["connection"]. Two reasons, both seen on a real service:

A copy drifts. A service rewrote env.py by hand to pass lint, ended up with its own copy of the SDK's logic — later fixes never reached it — and lost the sys.path handling along the way, so uv run alembic downgrade -1 stopped importing src. With a call into the SDK, the fix arrives when the package is upgraded.
A full script cannot be lint-clean under every configuration. The alembic/ directory next to pyproject.toml makes ruff's default (src = [".", "src"]) treat alembic as first-party, and the tempest new scaffold (src = ["src", "tests"]) as third-party. from alembic import context lands in a different import block under each, so one of the two always reports I001. The generated file imports no alembic, and sorts the same under both.

The guard tests/test_alembic_env_lint_guard.py generates the file in a temporary project and runs ruff check --select E,W,F,I under both configurations (plus the scaffold's whole configuration): zero errors and zero inline comments.

The bare Alembic CLI works too¶

The generated alembic.ini carries script_location = %(here)s/alembic and prepend_sys_path = %(here)s, both anchored on the ini's own directory. So uv run alembic upgrade head and uv run alembic downgrade -1 import your package without the tempest CLI, and a lifespan that passes the ini's absolute path (AlembicHelper("/app/alembic.ini")) finds the scripts whatever directory the process started in. Up to 0.312.x the ini carried alembic and ., relative to the working directory, and only worked from the project root. path_separator = os comes with it: without it, Alembic 1.19.1 emits a DeprecationWarning about the legacy splitting of prepend_sys_path, and anyone running with -W error (or filterwarnings = ["error"]) sees the migration fail.

One more hook¶

The hooks every revision goes through are in DEFAULT_REVISION_HOOKS. To add your own, compose and pass it as process_revision_directives= — the value replaces the default, so include the SDK's:

"""Alembic environment script with one extra revision hook."""

from typing import Any

from alembic.operations.ops import MigrationScript
from alembic.runtime.migration import MigrationContext
from tempest_fastapi_sdk.db import (
    DEFAULT_REVISION_HOOKS,
    compose_hooks,
    run_alembic_env,
)

from src.db.models import BaseModel


def skip_empty_revisions(
    context: MigrationContext,
    revision: Any,
    directives: list[Any],
) -> None:
    """Write no file when autogenerate finds no change."""
    script = directives[0]
    if (
        isinstance(script, MigrationScript)
        and script.upgrade_ops is not None
        and script.upgrade_ops.is_empty()
    ):
        directives[:] = []


run_alembic_env(
    BaseModel.metadata,
    process_revision_directives=compose_hooks(
        *DEFAULT_REVISION_HOOKS,
        skip_empty_revisions,
    ),
)

Measured against PostgreSQL 16: after upgrade head, an alembic revision --autogenerate with no model change writes no file.

Already have an env.py generated earlier?

It keeps working. To start receiving the SDK's fixes, replace its content with the three statements above (keeping your metadata import) and add prepend_sys_path = %(here)s and path_separator = os to the [alembic] section of alembic.ini. Migrations already written do not change: op.replace_enum and op.drop_enum_type are registered when env.py imports tempest_fastapi_sdk.db.

Each revision file¶

tempest db revision -m "Cria tabela de anotações" (and AlembicHelper.revision) writes:

alembic/versions/2026_10_11_2317-c16a7545f8f9_cria_tabela_de_anotacoes.py

An ASCII name. The message is folded to ASCII only to build the file name, so alembic/versions/ holds no accented module name; the migration's docstring starts with the message as you wrote it, accents included.
No comment line. tempest db init writes the SDK's own script.py.mako, without the # revision identifiers line of Alembic's template, and the revision drops the # ### commands auto generated by Alembic ### markers autogenerate puts around the operations.

With bare alembic revision

The new template applies (no # revision identifiers), but the name keeps its accents and the markers stay: measured with alembic revision --autogenerate -m "Cria anotações", the file came out as ..._cria_anotações.py with all four markers. Use tempest db revision.

Project initialized before this?

The ASCII name and the marker removal already apply, because they live in tempest db revision. For the comment-free template, replace alembic/script.py.mako with the SDK's:

uv run python -c "from importlib import resources; print(resources.files('tempest_fastapi_sdk.db._alembic_templates').joinpath('script.py.mako').read_text())" > alembic/script.py.mako

Where `create_tables()` is legitimate¶

In tests and throwaway local development — where the database is born and dies in the same process, and there is no migration to drift from:

# tests/conftest.py
from collections.abc import AsyncIterator

import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from tempest_fastapi_sdk import AsyncDatabaseManager


@pytest.fixture
async def session() -> AsyncIterator[AsyncSession]:
    """Yield a session over a fresh in-memory schema."""
    db: AsyncDatabaseManager = AsyncDatabaseManager(
        "sqlite+aiosqlite:///:memory:"
    )
    await db.create_tables()
    async with db.get_session_context() as opened:
        yield opened
    await db.drop_tables()

The technical reason it works here and not in production is the same in both cases: create_all only knows how to create what is missing. In a database that was just born, "what is missing" is everything — so it gets it right. In one that has already run, "what is missing" is no table at all, but possibly several columns — and columns it does not look at.

Recap

sync_schema() is the whole bootstrap: it tells an empty database, a pre-Alembic one and an already-migrated one apart, and reports which path it took.
From async code (lifespan, endpoint), call the _async twin (await helper.sync_schema_async()); the sync method raises with a loop running, naming the _async one to use.
create_tables() is CREATE TABLE IF NOT EXISTS — a silent no-op on an existing table. It is never the step that evolves a schema.
When adopting an existing schema, stamp the base revision (helper.base_revision(), or just helper.adopt()), never head.
To repair a wrong stamp("head"): stamp("base"), then sync_schema().
create_tables() stays legitimate where there is no migration to drift from: tests and in-memory SQLite.
On SQLite, migrations run on a connection without foreign keys: batch mode rebuilds the table and the cascade would delete the children. The generated env.py refuses a handed-over connection with FK on.
The generated env.py is one call to run_alembic_env: SDK fixes arrive with the package upgrade, and the file passes ruff with no inline comment. prepend_sys_path = %(here)s lets the bare Alembic CLI import your package.

Migrations¶

The complete bootstrap¶

From async code, use the _async method¶

The three states it tells apart¶

Why create_tables() is not the answer¶