feat(fm): replace create_all with alembic migrations

# Alembic configuration. The database URL is not set here: env.py reads it from

# Settings.postgres_url so migrations and the app never disagree about the target.

[alembic]

script_location = src/federation_manager/migrations

prepend_sys_path = src

version_path_separator = os

[loggers]

keys = root,sqlalchemy,alembic

[handlers]

keys = console

[formatters]

keys = generic

[logger_root]

level = WARNING

handlers = console

qualname =

[logger_sqlalchemy]

level = WARNING

handlers =

qualname = sqlalchemy.engine

[logger_alembic]

level = INFO

handlers =

qualname = alembic

[handler_console]

class = StreamHandler

args = (sys.stderr,)

level = NOTSET

formatter = generic

[formatter_generic]

format = %(levelname)-5.5s [%(name)s] %(message)s

datefmt = %H:%M:%S

readme = "README.md"

requires-python = ">=3.12"

dependencies = [

    "alembic>=1.13",

    "asyncpg>=0.30",

    "fastapi[standard]>=0.115",

    "httpx>=0.27",

where = ["src"]

include = ["federation_manager*"]

[tool.setuptools.package-data]

# Migrations ship with the package so the container image, which copies only src/,

# can run them at startup.

federation_manager = ["migrations/*.py", "migrations/*.mako", "migrations/versions/*.py"]

[tool.ruff]

line-length = 100

target-version = "py312"

from pathlib import Path

import structlog

from alembic import command

from alembic.config import Config

from sqlalchemy import inspect

from sqlalchemy.engine import Connection

from sqlalchemy.ext.asyncio import (

    AsyncEngine,

    AsyncSession,

    create_async_engine,

)

from federation_manager.adapters.database.tables import metadata

logger: structlog.BoundLogger = structlog.get_logger(__name__)

BASELINE_REVISION = "0001_baseline"

def build_engine(url: str, echo: bool = False) -> AsyncEngine:

    return async_sessionmaker(engine, expire_on_commit=False)

async def create_schema(engine: AsyncEngine) -> None:

def _alembic_config() -> Config:

    """Config built in code, not from alembic.ini.

    The migrations ship inside the package, so they are present in the container

    image, which copies only src/. alembic.ini exists for the developer CLI.

    """

    migrations = Path(__file__).resolve().parents[2] / "migrations"

    config = Config()

    config.set_main_option("script_location", str(migrations))

    return config

def _needs_baseline_stamp(connection: Connection) -> bool:

    """True for a database built by the old create_all path.

    Those have the tables but no alembic_version, so an upgrade would try to create

    what is already there. Stamping adopts them once; afterwards this is never true.

    """

    tables = set(inspect(connection).get_table_names())

    return "alembic_version" not in tables and "partner_ops" in tables

def _upgrade(connection: Connection) -> None:

    config = _alembic_config()

    config.attributes["connection"] = connection

    if _needs_baseline_stamp(connection):

        logger.warning("adopting_pre_alembic_database", stamped_as=BASELINE_REVISION)

        command.stamp(config, BASELINE_REVISION)

    command.upgrade(config, "head")

async def run_migrations(engine: AsyncEngine) -> None:

    """Bring fm_db to head.

    Replaces metadata.create_all, which only ever created missing tables and so could

    not add a column to a database that already existed.

    """

    async with engine.begin() as conn:

        await conn.run_sync(metadata.create_all)

        await conn.run_sync(_upgrade)

from federation_manager.adapters.database.core import (

    build_engine,

    build_session_maker,

    create_schema,

    run_migrations,

)

from federation_manager.adapters.database.federation_context_repo import (

    PostgresFederationContextRepo,

    settings = get_settings()

    configure_logging(settings.log_level)

    engine = build_engine(settings.postgres_url, echo=settings.postgres_echo)

    await create_schema(engine)

    await run_migrations(engine)

    app.state.session_maker = build_session_maker(engine)

    app.state.jwt_validator = KeycloakJwtValidator(settings.keycloak_issuer)

    publisher = NatsCommandPublisher(settings.nats_url)

"""Alembic environment.

The URL comes from Settings, not alembic.ini, so a migration run can never target a

different database than the application.

"""

import asyncio

from logging.config import fileConfig

from alembic import context

from sqlalchemy.engine import Connection

from sqlalchemy.ext.asyncio import async_engine_from_config

from sqlalchemy.pool import NullPool

from federation_manager.adapters.database.tables import metadata

from federation_manager.core.config import get_settings

config = context.config

if config.config_file_name is not None:

    fileConfig(config.config_file_name)

target_metadata = metadata

config.set_main_option("sqlalchemy.url", get_settings().postgres_url)

def _configure(connection: Connection) -> None:

    context.configure(

        connection=connection,

        target_metadata=target_metadata,

        # Without this, autogenerate produces no diff for a column whose type or

        # nullability changed, which is most of what we will be doing.

        compare_type=True,

        compare_server_default=True,

    )

def run_migrations_offline() -> None:

    context.configure(

        url=config.get_main_option("sqlalchemy.url"),

        target_metadata=target_metadata,

        literal_binds=True,

        dialect_opts={"paramstyle": "named"},

        compare_type=True,

        compare_server_default=True,

    )

    with context.begin_transaction():

        context.run_migrations()

def _run(connection: Connection) -> None:

    _configure(connection)

    with context.begin_transaction():

        context.run_migrations()

async def run_migrations_online() -> None:

    engine = async_engine_from_config(

        config.get_section(config.config_ini_section, {}),

        prefix="sqlalchemy.",

        poolclass=NullPool,

    )

    async with engine.connect() as connection:

        await connection.run_sync(_run)

    await engine.dispose()

if context.is_offline_mode():

    run_migrations_offline()

elif (connection := config.attributes.get("connection")) is not None:

    # Called in-process by the app, which already holds a connection.

    _run(connection)

else:

    asyncio.run(run_migrations_online())