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.
migrations_path str "" Alembic script directory (the one containing env.py). Empty disables startup migrations. Requires pico-sqlalchemy[migrations].
migrations_target str "head" Alembic revision to upgrade to on startup.

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_boot import init
from pico_ioc import configuration, DictSource

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

container = init(modules=["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.

Any @component with a configure_database(engine) method is collected and called during application startup — subclassing this protocol is optional. Hooks always run OFF the event loop (on a worker thread when the container boots under an ASGI server), so the asyncio.run() pattern below is safe in every context.

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

An optional priority attribute (int, default 0) controls the execution order among multiple configurers; lower values run first.

Example::

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

    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)
            await engine.dispose()  # asyncpg pools are loop-bound
        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
@runtime_checkable
class DatabaseConfigurer(Protocol):
    """Protocol for database startup hooks.

    Any ``@component`` with a ``configure_database(engine)`` method is
    collected and called during application startup — subclassing this
    protocol is optional. Hooks always run OFF the event loop (on a worker
    thread when the container boots under an ASGI server), so the
    ``asyncio.run()`` pattern below is safe in every context.

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

    An optional ``priority`` attribute (int, default ``0``) controls the
    execution order among multiple configurers; lower values run first.

    Example::

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

            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)
                    await engine.dispose()  # asyncpg pools are loop-bound
                asyncio.run(_create())
    """

    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 
47
48
49
50
51
52
53
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.
migrations_path str

Alembic script directory (the one containing env.py). Empty string (default) disables startup migrations. Requires pico-sqlalchemy[migrations].
migrations_target str

Alembic revision to upgrade to on startup.
Source code in src/pico_sqlalchemy/config.py 
56
57
58
59
60
61
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.
        migrations_path: Alembic script directory (the one containing
            ``env.py``). Empty string (default) disables startup
            migrations. Requires ``pico-sqlalchemy[migrations]``.
        migrations_target: Alembic revision to upgrade to on startup.
    """

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