Merge branch 'feat/alembic-migrations' into 'develop'

# 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)

            )

        )

    async def find_outbound(

        self, partner_id: UUID, federation_context_id: str

    ) -> FederationContext | None:

        return await self._one(

            select(contexts)

            .where(

                contexts.c.partner_op_id == partner_id,

                contexts.c.direction == "outbound",

                contexts.c.federation_context_id == federation_context_id,

            )

            .order_by(contexts.c.created_at.desc())

            .limit(1)

        )

    async def add(self, context: FederationContext) -> None:

        await self._session.execute(

            insert(contexts).values(

        self._allow_insecure = allow_insecure

    async def post(self, partner: PartnerOP, path: str, payload: dict[str, object]) -> EwbiResponse:

        return await self._send(partner, "POST", path, payload=payload)

    async def get(

        self, partner: PartnerOP, path: str, params: dict[str, str] | None = None

    ) -> EwbiResponse:

        return await self._send(partner, "GET", path, params=params)

    async def delete(self, partner: PartnerOP, path: str) -> EwbiResponse:

        return await self._send(partner, "DELETE", path)

    async def _send(

        self,

        partner: PartnerOP,

        method: str,

        path: str,

        *,

        payload: dict[str, object] | None = None,

        params: dict[str, str] | None = None,

    ) -> EwbiResponse:

        url = self._url(partner, path)

        token = await self._token_provider.token_for(partner, scope="fed-mgmt")

        try:

            response = await self._client.post(

            response = await self._client.request(

                method,

                url,

                headers={"Accept": "application/json", "Authorization": f"Bearer {token}"},

                json=payload,

                params=params,

            )

        except httpx.HTTPError:

            raise PartnerRequestFailed(partner.id) from None