Skip to content

Database Configuration

This reference covers the database configuration extension points: the DatabaseSettings dataclass and the DatabaseConfigurer protocol. Together, they provide a consistent way to define default database connection settings and to hook into the database engine initialization to apply project-specific configuration.

What is this?

  • DatabaseSettings (dataclass)
  • A @configured dataclass that holds the database connection settings.
  • Automatically loaded from configuration sources (YAML, env, dict) via the database prefix.

  • Acts as the single source of truth for connection parameters (URL, pooling, echo).

  • DatabaseConfigurer (protocol)

  • A small extension point for customizing the database engine after it is created.
  • Members:
    • priority: int attribute (or property) used to order multiple configurers. Lower numbers run first.
    • configure_database(self, engine): Applies configuration to the database engine.

1. DatabaseSettings

The DatabaseSettings class is a @configured dataclass. It is automatically populated from your configuration source and injected where needed by the container.

Supported Fields

The following fields map directly to the underlying SQLAlchemy create_async_engine parameters:

Field Type Default Description
url str "sqlite+aiosqlite:///./app.db" The SQLAlchemy connection URL (DSN). Must use an async driver (e.g., postgresql+asyncpg).
echo bool False If True, the engine will log all SQL statements to stdout.
pool_size int 5 The number of connections to keep open inside the connection pool.
pool_pre_ping bool True If True, tests connections for liveness upon checkout (prevents "MySQL server has gone away" errors).
pool_recycle int 3600 Recycle connections after this many seconds to prevent timeouts from the database side.

Note: pool_size, pool_pre_ping, and pool_recycle are ignored if using SQLite with :memory:.

Providing Settings

Settings are loaded automatically from a configuration source with the database prefix:

from pico_ioc import init, configuration, DictSource

config = configuration(DictSource({
    "database": {
        "url": "postgresql+asyncpg://user:pass@localhost:5432/dbname",
        "echo": True,
        "pool_size": 10,
    }
}))

container = init(modules=["pico_sqlalchemy", "myapp"], config=config)

Or via YAML:

# application.yaml
database:
  url: postgresql+asyncpg://user:pass@localhost:5432/dbname
  echo: false
  pool_size: 10
  pool_pre_ping: true
  pool_recycle: 3600

2. DatabaseConfigurer

Create one or more classes that implement the DatabaseConfigurer protocol to apply engine-level behavior customization.

Ordering: Configurers are executed in ascending order based on their priority attribute. Lower numbers run first.

Example: Enable SQL Echo (Dynamic)

from pico_ioc import component
from pico_sqlalchemy import DatabaseConfigurer

@component
class EnableSqlEcho(DatabaseConfigurer):
    priority = 10  # Runs early

    def configure_database(self, engine):
        engine.echo = True

Example: SQLite Pragmas

from sqlalchemy import event
from pico_ioc import component
from pico_sqlalchemy import DatabaseConfigurer

@component
class SQLitePragmaConfigurer(DatabaseConfigurer):
    priority = 50  # Runs after basic setup

    def configure_database(self, engine):
        @event.listens_for(engine.sync_engine, "connect")
        def set_sqlite_pragma(dbapi_connection, connection_record):
            cursor = dbapi_connection.cursor()
            cursor.execute("PRAGMA foreign_keys=ON")
            cursor.close()

Example: Create Tables on Startup

import asyncio
from pico_ioc import component
from pico_sqlalchemy import DatabaseConfigurer, AppBase

@component
class TableCreationConfigurer(DatabaseConfigurer):
    priority = 100

    def __init__(self, base: AppBase):
        self.base = base

    def configure_database(self, engine):
        async def init_schema():
            async with engine.begin() as conn:
                await conn.run_sync(self.base.metadata.create_all)
        asyncio.run(init_schema())

3. Initialization Logic

The startup sequence involves two separate components:

  1. SqlAlchemyFactory creates the SessionManager singleton from DatabaseSettings (creates the AsyncEngine and session factory).
  2. PicoSqlAlchemyLifecycle (via @configure) collects all DatabaseConfigurer implementations, sorts them by priority (ascending), and calls configure_database(engine) on each.

This ensures that your specific database tuning (like pragmas or connection pool listeners) is applied reliably before the application starts accepting traffic.

Auto-generated API

pico_sqlalchemy.config

Database configuration types for pico-sqlalchemy.

  • DatabaseSettings -- a @configured dataclass that is automatically populated from the "database" configuration prefix.
  • DatabaseConfigurer -- a protocol for hooks that run against the AsyncEngine during startup (e.g. table creation, migrations).

DatabaseConfigurer

Bases: Protocol

Protocol for database startup hooks.

Implement this protocol and register the class as a @component to have configure_database() called automatically during application startup, in ascending priority order.

Typical uses include DDL creation, Alembic migrations, and seed data loading.

Attributes:

Name Type Description
priority int

Controls the execution order among multiple configurers. Lower values run first. Defaults to 0.

Example::

@component
class SchemaSetup(DatabaseConfigurer):
    def __init__(self, base: AppBase):
        self.base = base

    @property
    def priority(self) -> int:
        return 0

    def configure_database(self, engine) -> None:
        import asyncio
        async def _create():
            async with engine.begin() as conn:
                await conn.run_sync(self.base.metadata.create_all)
        asyncio.run(_create())
Source code in src/pico_sqlalchemy/config.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@runtime_checkable
class DatabaseConfigurer(Protocol):
    """Protocol for database startup hooks.

    Implement this protocol and register the class as a ``@component``
    to have ``configure_database()`` called automatically during application
    startup, in ascending ``priority`` order.

    Typical uses include DDL creation, Alembic migrations, and seed
    data loading.

    Attributes:
        priority: Controls the execution order among multiple
            configurers.  Lower values run first.  Defaults to ``0``.

    Example::

        @component
        class SchemaSetup(DatabaseConfigurer):
            def __init__(self, base: AppBase):
                self.base = base

            @property
            def priority(self) -> int:
                return 0

            def configure_database(self, engine) -> None:
                import asyncio
                async def _create():
                    async with engine.begin() as conn:
                        await conn.run_sync(self.base.metadata.create_all)
                asyncio.run(_create())
    """

    @property
    def priority(self) -> int:
        return 0

    def configure_database(self, engine) -> None:
        """Run configuration against the given ``AsyncEngine``.

        Args:
            engine: The ``AsyncEngine`` created by ``SessionManager``.
        """
        raise NotImplementedError

configure_database(engine)

Run configuration against the given AsyncEngine.

Parameters:

Name Type Description Default
engine

The AsyncEngine created by SessionManager.

 required
Source code in src/pico_sqlalchemy/config.py 
53
54
55
56
57
58
59
def configure_database(self, engine) -> None:
    """Run configuration against the given ``AsyncEngine``.

    Args:
        engine: The ``AsyncEngine`` created by ``SessionManager``.
    """
    raise NotImplementedError

DatabaseSettings dataclass

Type-safe database connection settings.

Populated automatically from the "database" configuration prefix via pico-ioc's @configured mechanism. For example, with a DictSource::

configuration(DictSource({
    "database": {
        "url": "postgresql+asyncpg://user:pass@host/db",
        "echo": False,
        "pool_size": 10,
    }
}))

Attributes:

Name Type Description
url str

SQLAlchemy connection URL (default: in-memory SQLite).
echo bool

Log all SQL statements if True.
pool_size int

Connection pool size.
pool_pre_ping bool

Test connections before checkout.
pool_recycle int

Recycle connections after this many seconds.
Source code in src/pico_sqlalchemy/config.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@configured(target="self", prefix="database", mapping="tree")
@dataclass
class DatabaseSettings:
    """Type-safe database connection settings.

    Populated automatically from the ``"database"`` configuration prefix
    via pico-ioc's ``@configured`` mechanism.  For example, with a
    ``DictSource``::

        configuration(DictSource({
            "database": {
                "url": "postgresql+asyncpg://user:pass@host/db",
                "echo": False,
                "pool_size": 10,
            }
        }))

    Attributes:
        url: SQLAlchemy connection URL (default: in-memory SQLite).
        echo: Log all SQL statements if ``True``.
        pool_size: Connection pool size.
        pool_pre_ping: Test connections before checkout.
        pool_recycle: Recycle connections after this many seconds.
    """

    url: str = "sqlite+aiosqlite:///./app.db"
    echo: bool = False
    pool_size: int = 5
    pool_pre_ping: bool = True
    pool_recycle: int = 3600