sqlalchemy-foundation-kit

Tip

Building this with an AI assistant? Hand it one page instead of the whole site: the public API surface, the session-lifetime and transaction-ownership rules that break code when they are broken, the mistakes models actually make with a Unit of Work, and a map of which page to fetch for the rest. Every docs page is also served as raw Markdown at its own URL, and a Copy page button at the top of each one hands it straight to a chat window.

Key Features

Installation

# Basic installation (core functionality)
pip install sqlalchemy-foundation-kit

# With pydantic-settings support
pip install sqlalchemy-foundation-kit[settings]

# With Prometheus metrics
pip install sqlalchemy-foundation-kit[metrics]

# With OpenTelemetry tracing
pip install sqlalchemy-foundation-kit[telemetry]

# With dishka dependency injection
pip install sqlalchemy-foundation-kit[dishka]

# With dependency-injector containers
pip install sqlalchemy-foundation-kit[dependency-injector]

# With orjson serialization
pip install sqlalchemy-foundation-kit[orjson]

# All features
pip install sqlalchemy-foundation-kit[all]

Quick Start

1. Configuration (with pydantic-settings)

from pydantic import SecretStr
from pydantic_settings import BaseSettings
from sqlalchemy_foundation_kit.contrib.settings import (
    BasePostgresConfig,
    ConnectionSettings,
    PoolSettings,
    QuerySettings,
)

class Settings(BaseSettings):
    postgres: BasePostgresConfig = BasePostgresConfig(
        connection=ConnectionSettings(
            host="localhost",
            port=5432,
            user="postgres",
            password=SecretStr("secret"),
            database="mydb",
        ),
        pool=PoolSettings(),  # Uses defaults
        query=QuerySettings(),  # Uses defaults
        application_name="my-service",
    )

from dataclasses import dataclass
from sqlalchemy_foundation_kit import (
    PostgresSettingsProtocol,
    ConnectionSettingsProtocol,
    PoolSettingsProtocol,
    QuerySettingsProtocol,
)

@dataclass
class MyConnectionSettings:
    host: str = "localhost"
    port: int = 5432
    user: str = "postgres"
    database: str = "mydb"

@dataclass
class MyPoolSettings:
    kind: str = "async_adapted_queue"
    size: int | None = 10
    max_overflow: int | None = 20
    pre_ping: bool = True
    recycle: int | None = 3600
    timeout: float | None = 30.0

@dataclass
class MyQuerySettings:
    echo: bool = False
    statement_cache_size: int | None = 0
    prepared_statement_cache_size: int | None = 0
    isolation_level: str | None = None

class PostgresConfig:
    connection: ConnectionSettingsProtocol = MyConnectionSettings()
    pool: PoolSettingsProtocol = MyPoolSettings()
    query: QuerySettingsProtocol = MyQuerySettings()
    application_name: str = "my-service"
    db_schema: str | None = None
    use_orjson_serialization: bool = False
    jit: str | None = None
    
    def to_dsn(self) -> str:
        return f"postgresql+asyncpg://{self.connection.user}@{self.connection.host}:{self.connection.port}/{self.connection.database}"

2. Define ORM Models

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy_foundation_kit import BaseTable, DatetimeColumnsMixin

class UserDB(BaseTable, DatetimeColumnsMixin):
    __tablename__ = "users"
    __created_at_index__ = True  # Index on created_at

    id: Mapped[UUID] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    username: Mapped[str]

3. Create Session Manager

from sqlalchemy_foundation_kit import create_async_session_manager

async def main():
    session_manager = create_async_session_manager(
        settings.postgres,
        metrics=postgres_metrics,  # optional
    )
    
    async with session_manager.get_transaction() as session:
        user = UserDB(id=uuid4(), email="[email protected]", username="user")
        session.add(user)
        # Auto-commit on exit

    await session_manager.aclose()

4. Unit of Work Pattern

from sqlalchemy_foundation_kit import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction

# Define your transaction with repositories
class MyTransaction(AsyncSQLAlchemyUowTransaction):
    @property
    def users(self) -> UserRepository:
        if self._users is None:
            self._users = PostgresUserRepository(self.session)
        return self._users

# Create UoW
class MyUnitOfWork(AsyncSQLAlchemyUnitOfWork[MyTransaction]):
    def __init__(self, session_maker):
        super().__init__(session_maker, transaction_factory=MyTransaction)

# Usage in use case
class CreateUserUseCase:
    def __init__(self, uow: MyUnitOfWork):
        self._uow = uow
    
    async def execute(self, email: str) -> User:
        async with self._uow.transaction() as tx:
            user = await tx.users.create(email=email)
            await tx.outbox.create(UserCreatedEvent(...))
        return user

What's Included

Core (always available)

Contrib (optional dependencies)

contrib.settings (requires [settings])

from sqlalchemy_foundation_kit.contrib.settings import (
    BasePostgresConfig,
    BasePostgresMigrationsConfig,
)

contrib.metrics (requires [metrics])

from sqlalchemy_foundation_kit.contrib.metrics import PostgresMetrics

contrib.di (requires [dishka])

from sqlalchemy_foundation_kit.contrib.di import (
    AsyncDatabaseProvider,
    AsyncUnitOfWorkProvider,
    PrometheusPostgresMetricsProvider,
)

from dishka import make_async_container
from sqlalchemy_foundation_kit.contrib.di import (
    AsyncDatabaseProvider,
    AsyncUnitOfWorkProvider,
    PrometheusPostgresMetricsProvider,
)

container = make_async_container(
    AsyncDatabaseProvider(),
    AsyncUnitOfWorkProvider(),
    PrometheusPostgresMetricsProvider(),
    # ... your providers, which provide the settings these ask for:
    # see "Dependency Injection" in the advanced guide
)

contrib.dependency_injector (requires [dependency-injector])

from sqlalchemy_foundation_kit.contrib.dependency_injector import (
    DatabaseContainer,
    AsyncDatabaseResourceProvider,
    PrometheusMetricsContainer,
)

from dependency_injector import containers, providers
from sqlalchemy_foundation_kit.contrib.dependency_injector import (
    DatabaseContainer,
    PrometheusMetricsContainer,
)

class AppContainer(containers.DeclarativeContainer):
    config = providers.Singleton(Settings)
    
    metrics = providers.Container(
        PrometheusMetricsContainer,
        metrics_settings=config.provided.metrics,
        default_prefix=providers.Object("myapp"),
        postgres_settings=config.provided.postgres,
    )
    
    database = providers.Container(
        DatabaseContainer,
        postgres_config=config.provided.postgres,
        metrics=metrics.postgres_metrics,
    )

container = AppContainer()
await container.init_resources()

# Use dependencies
uow = container.database.uow()
async with uow.transaction() as tx:
    user = await tx.users.create(...)

contrib.telemetry (requires [telemetry])

from sqlalchemy_foundation_kit.contrib.telemetry import (
    instrument_sqlalchemy,
    instrument_asyncpg,
    TracedAsyncUnitOfWork,
)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, BatchSpanProcessor

# Setup tracer
provider = TracerProvider()
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Instrument SQLAlchemy
from sqlalchemy_foundation_kit.contrib.telemetry import (
    instrument_sqlalchemy,
    TracedAsyncUnitOfWork,
)

instrument_sqlalchemy(engine=engine)

# Use traced UoW
uow = TracedAsyncUnitOfWork(
    session_maker=session_maker,
    transaction_factory=MyTransaction,
    service_name="my-service",
)

async with uow.transaction() as tx:
    # Automatically creates span "uow.transaction"
    user = await tx.users.create(...)

Architecture

Documentation

License