feat(fastapi): refactored integration with CLI support (#352)

Refactored the Starlette and FastAPI integration to support multiple configurations and sessions.  Additionally, there's now a method for attaching the Advanced Alchemy database CLI group to the FastAPI CLI.

advanced_alchemy/extensions/fastapi/__init__.py

@@ -0,0 +1,36 @@
+"""FastAPI extension for Advanced Alchemy.
+
+This module provides FastAPI integration for Advanced Alchemy, including session management,
+database migrations, and service utilities.
+
+"""
+
+from advanced_alchemy import base, exceptions, filters, mixins, operations, repository, service, types, utils
+from advanced_alchemy.alembic.commands import AlembicCommands
+from advanced_alchemy.config import AlembicAsyncConfig, AlembicSyncConfig, AsyncSessionConfig, SyncSessionConfig
+from advanced_alchemy.extensions.fastapi.cli import get_database_migration_plugin
+from advanced_alchemy.extensions.fastapi.config import EngineConfig, SQLAlchemyAsyncConfig, SQLAlchemySyncConfig
+from advanced_alchemy.extensions.fastapi.extension import AdvancedAlchemy, assign_cli_group
+
+__all__ = (
+    "AdvancedAlchemy",
+    "AlembicAsyncConfig",
+    "AlembicCommands",
+    "AlembicSyncConfig",
+    "AsyncSessionConfig",
+    "EngineConfig",
+    "SQLAlchemyAsyncConfig",
+    "SQLAlchemySyncConfig",
+    "SyncSessionConfig",
+    "assign_cli_group",
+    "base",
+    "exceptions",
+    "filters",
+    "get_database_migration_plugin",
+    "mixins",
+    "operations",
+    "repository",
+    "service",
+    "types",
+    "utils",
+)

advanced_alchemy/extensions/fastapi/cli.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional, cast
+
+try:
+    import rich_click as click
+except ImportError:
+    import click  # type: ignore[no-redef]
+
+from advanced_alchemy.cli import add_migration_commands
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
+    from advanced_alchemy.extensions.fastapi.extension import AdvancedAlchemy
+
+
+def get_database_migration_plugin(app: FastAPI) -> AdvancedAlchemy:  # pragma: no cover
+    """Retrieve the Advanced Alchemy extension from a FastAPI application instance."""
+    from advanced_alchemy.exceptions import ImproperConfigurationError
+
+    extension = cast("Optional[AdvancedAlchemy]", getattr(app.state, "advanced_alchemy", None))
+    if extension is None:
+        msg = "Failed to initialize database CLI. The Advanced Alchemy extension is not properly configured."
+        raise ImproperConfigurationError(msg)
+    return extension
+
+
+def register_database_commands(app: FastAPI) -> click.Group:  # pragma: no cover
+    @click.group(name="database")
+    @click.pass_context
+    def database_group(ctx: click.Context) -> None:
+        """Manage SQLAlchemy database components."""
+        ctx.ensure_object(dict)
+        ctx.obj["configs"] = get_database_migration_plugin(app).config
+
+    add_migration_commands(database_group)
+    return database_group

advanced_alchemy/extensions/fastapi/config.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from advanced_alchemy.extensions.starlette import EngineConfig, SQLAlchemyAsyncConfig, SQLAlchemySyncConfig
+
+__all__ = (
+    "EngineConfig",
+    "SQLAlchemyAsyncConfig",
+    "SQLAlchemySyncConfig",
+)

advanced_alchemy/extensions/fastapi/extension.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Sequence
+
+from advanced_alchemy.extensions.fastapi.cli import register_database_commands
+from advanced_alchemy.extensions.starlette import AdvancedAlchemy as StarletteAdvancedAlchemy
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
+    from advanced_alchemy.extensions.fastapi.config import SQLAlchemyAsyncConfig, SQLAlchemySyncConfig
+
+__all__ = ("AdvancedAlchemy",)
+
+
+def assign_cli_group(app: FastAPI) -> None:  # pragma: no cover
+    try:
+        from fastapi_cli.cli import app as fastapi_cli_app  # pyright: ignore[reportUnknownVariableType]
+        from typer.main import get_group
+    except ImportError:
+        print("FastAPI CLI is not installed.  Skipping CLI registration.")  # noqa: T201
+        return
+    click_app = get_group(fastapi_cli_app)  # pyright: ignore[reportUnknownArgumentType]
+    click_app.add_command(register_database_commands(app))
+
+
+class AdvancedAlchemy(StarletteAdvancedAlchemy):
+    """AdvancedAlchemy integration for FastAPI applications.
+
+    This class manages SQLAlchemy sessions and engine lifecycle within a FastAPI application.
+    It provides middleware for handling transactions based on commit strategies.
+    """
+
+    def __init__(
+        self,
+        config: SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig],
+        app: FastAPI | None = None,
+    ) -> None:
+        super().__init__(config, app)