Configuration¶

sqlalchemy-foundation-kit provides flexible configuration options through protocols and optional Pydantic-based settings.

Configuration Approaches¶

You have two options for configuring the library:

1. contrib.settings — Pydantic-based configuration (requires [settings] extra)
2. Protocol-based — Implement PostgresSettingsProtocol directly

Using contrib.settings (Recommended)¶

Install with settings support:

pip install sqlalchemy-foundation-kit[settings]

Basic Configuration¶

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(
            size=10,
            max_overflow=20,
            timeout=30.0,
            pre_ping=True,
            recycle=3600,
        ),
        query=QuerySettings(
            echo=False,
            statement_cache_size=0,
            prepared_statement_cache_size=0,
            isolation_level="READ COMMITTED",
        ),
        application_name="my-service",
        db_schema=None,
        use_orjson_serialization=True,
        jit=None,
        metrics_enabled=True,
    )

settings = Settings()

Environment Variables¶

BasePostgresConfig declares no model_config of its own, so it reads no prefix and no delimiter by itself. Environment variables reach it through the BaseSettings that holds it, which is where env_nested_delimiter is set:

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_nested_delimiter="__")

    postgres: BasePostgresConfig

Every level of nesting is then one __, the field name included — postgres is a field like any other, so postgres.connection.host is POSTGRES__CONNECTION__HOST:

# Connection settings
POSTGRES__CONNECTION__HOST=db.example.com
POSTGRES__CONNECTION__PORT=5432
POSTGRES__CONNECTION__USER=postgres
POSTGRES__CONNECTION__PASSWORD=secret123
POSTGRES__CONNECTION__DATABASE=mydb

# Pool settings
POSTGRES__POOL__SIZE=20
POSTGRES__POOL__MAX_OVERFLOW=30
POSTGRES__POOL__TIMEOUT=45.0
POSTGRES__POOL__PRE_PING=true
POSTGRES__POOL__RECYCLE=1800

# Query settings
POSTGRES__QUERY__ECHO=false
POSTGRES__QUERY__STATEMENT_CACHE_SIZE=0
POSTGRES__QUERY__ISOLATION_LEVEL="READ COMMITTED"

# Top-level settings
POSTGRES__APPLICATION_NAME=my-service
POSTGRES__DB_SCHEMA=public
POSTGRES__USE_ORJSON_SERIALIZATION=true
POSTGRES__JIT=off   # a startup parameter: direct connections only, see PgBouncer below
POSTGRES__METRICS_ENABLED=true

BasePostgresMigrationsConfig is exactly this settings class, ready made: it holds a postgres: BasePostgresConfig with env_nested_delimiter="__" and extra="ignore" already set, and reads the same names.

Custom prefix:

from pydantic_settings import SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="MY_APP_",
        env_nested_delimiter="__",
    )

    postgres: BasePostgresConfig

Now use MY_APP_POSTGRES__CONNECTION__HOST instead of POSTGRES__CONNECTION__HOST.

DSN Generation¶

# Generate DSN for asyncpg
dsn = settings.postgres.to_dsn()
# postgresql+asyncpg://postgres:secret@localhost:5432/mydb

# Generate DSN without driver
dsn = settings.postgres.to_dsn(driver=None)
# postgresql://postgres:secret@localhost:5432/mydb

# Generate DSN with masked password (for logging)
dsn = settings.postgres.to_dsn(mask_password=True)
# postgresql+asyncpg://postgres:**********@localhost:5432/mydb

Configuration Options¶

Connection Settings¶

Example:

ConnectionSettings(
    host="db.prod.example.com",
    port=5432,
    user="app_user",
    password=SecretStr("$ecr3t!"),
    database="production_db",
)

Pool Settings¶

Pool Types:

• async_adapted_queue (default) — Standard async queue-based pool, recommended for most use cases. With max_overflow=0 it is a fixed-size pool
• static — SQLAlchemy's StaticPool: exactly one connection, shared by every session. Concurrent sessions on it fail (another operation is in progress), so it is for tests, never for throughput. size must be 1
• null — No pooling, creates new connection for each request (not recommended for production)

size, max_overflow and timeout describe a queue of connections, so only the queue pools receive them; static and null ignore them. recycle and pre_ping apply to every pool.

Example:

# Development: small pool with quick recycling
PoolSettings(
    size=5,
    max_overflow=10,
    pre_ping=True,
    recycle=600,  # 10 minutes
    timeout=10.0,
)

# Production: large pool with longer recycling
PoolSettings(
    size=20,
    max_overflow=30,
    pre_ping=True,
    recycle=3600,  # 1 hour
    timeout=30.0,
)

# Fixed size: exactly 50 connections, never more
PoolSettings(
    size=50,
    max_overflow=0,
)

Validation:

The library validates pool settings:

# ❌ This raises ValueError: a static pool is one connection
PoolSettings(
    kind="static",
    size=50,
)

# ✅ Correct: one shared connection, for tests
PoolSettings(
    kind="static",
    size=1,
)

Query Settings¶

Isolation Levels:

• "READ UNCOMMITTED" — Lowest isolation, allows dirty reads
• "READ COMMITTED" — Default PostgreSQL level, prevents dirty reads
• "REPEATABLE READ" — Prevents non-repeatable reads
• "SERIALIZABLE" — Highest isolation, full transactional consistency

Example:

# Development: verbose logging
QuerySettings(
    echo=True,  # Log all SQL
    statement_cache_size=0,  # Disable caching for debugging
    isolation_level=None,  # Use database default
)

# Production: optimized for performance
QuerySettings(
    echo=False,  # No SQL logging
    statement_cache_size=500,  # Cache common queries
    prepared_statement_cache_size=500,
    isolation_level="READ COMMITTED",
)

# High-consistency workload
QuerySettings(
    echo=False,
    statement_cache_size=0,  # Disable for pgbouncer transaction mode
    prepared_statement_cache_size=0,
    isolation_level="SERIALIZABLE",  # Strongest guarantees
)

Top-Level Settings¶

Application Name:

The application_name appears in PostgreSQL logs and monitoring tools, helping identify which service is issuing queries:

BasePostgresConfig(
    application_name="identity-service",
    # ...
)

PostgreSQL pg_stat_activity will show:

SELECT application_name, query FROM pg_stat_activity;
-- application_name | query
-- identity-service | SELECT * FROM users WHERE id = $1

Schema:

BasePostgresConfig(
    db_schema="tenant_123",
    # ...
)

The value is a search_path, so "tenant_123, public" is valid too. It is applied at the start of every transaction with SET LOCAL semantics — set_config('search_path', …, true) as the first statement after BEGIN — which is the one scope that survives a transaction-mode pooler. Every transaction(), query(), managed_session(), get_session(), get_transaction() and raw engine.connect() block sees it, including the transaction that follows a commit() in the same session. Two things it is not: a startup parameter, which PgBouncer rejects or drops (see PgBouncer), and a session-level SET, which leaks between clients through a pooler. A statement run with isolation_level="AUTOCOMMIT" begins no transaction and therefore runs with the server's default search_path.

If you would rather not pay one round-trip per transaction, set it on the server instead — ALTER ROLE app_user SET search_path = tenant_123 or ALTER DATABASE mydb SET search_path = tenant_123 — and leave db_schema unset; or qualify the schema in your metadata (__table_args__ = {"schema": "tenant_123"}) and need no search_path at all. Outside create_async_session_manager the same mechanism is AsyncSessionManager(..., search_path="tenant_123") or AsyncSessionManagerBuilder(url).with_search_path("tenant_123").

orjson Serialization:

Off by default. Install [orjson] and turn it on for faster JSON serialization:

pip install sqlalchemy-foundation-kit[orjson]

BasePostgresConfig(
    use_orjson_serialization=True,
    # ...
)

It becomes the engine's JSON serializer, so every JSON / JSONB column uses it, PydanticJSONB included. orjson also serializes values json.dumps rejects — datetime, UUID, Decimal, non-str keys — so a service that turns it off again has to keep its JSON columns to plain JSON types. Up to 0.4.0 it was on by default.

JIT (Just-In-Time Compilation):

jit is None by default: nothing is sent and PostgreSQL's own setting applies (on since PostgreSQL 12). Set "off" or "on" only to override the server for this application, and know that the value travels as a startup parameter — fine on a direct connection, rejected by PgBouncer in transaction mode unless it is listed in track_extra_parameters (see PgBouncer). Behind a pooler, set it on the server instead: ALTER ROLE app_user SET jit = off.

# Direct PostgreSQL connection: override the server setting for this application
BasePostgresConfig(
    jit="off",
    # ...
)

Protocol-Based Configuration¶

If you don't want pydantic-settings, implement PostgresSettingsProtocol:

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

@dataclass
class MyConnectionSettings:
    host: str = "localhost"
    port: int = 5432
    user: str = "postgres"
    password: SecretStr = SecretStr("secret")
    database: str = "mydb"

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

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

@dataclass
class MyPostgresConfig:
    connection: ConnectionSettingsProtocol
    pool: PoolSettingsProtocol
    query: QuerySettingsProtocol
    application_name: str = "my-service"
    db_schema: str | None = None
    use_orjson_serialization: bool = False
    jit: str | None = None
    metrics_enabled: bool = False

    def to_dsn(self, driver: str | None = "asyncpg", mask_password: bool = False) -> str:
        conn = self.connection
        password = "**********" if mask_password else conn.password.get_secret_value()
        scheme = f"postgresql+{driver}" if driver else "postgresql"
        return f"{scheme}://{conn.user}:{password}@{conn.host}:{conn.port}/{conn.database}"

PgBouncer¶

Measured on PostgreSQL 17 and PgBouncer 1.25 in transaction mode, PgBouncer at its defaults unless the row says otherwise: twenty clients, twenty distinct statements each, run twice, so every cache fills. The scripts are on issue #23.

Four things follow. The library is built around the first three; the fourth is a setting you choose.

Startup parameters do not pass through. asyncpg's server_settings are parameters of the PostgreSQL startup packet. PgBouncer forwards only the ones it tracks — client_encoding, datestyle, timezone, standard_conforming_strings, application_name, plus whatever you list in track_extra_parameters (1.18+) — and refuses the connection on any other. ignore_startup_parameters makes the connection succeed by throwing the parameter away, which is worse: a search_path the library accepted and the database never saw. So create_async_session_manager sends application_name and nothing else by default; jit goes only when you set it, and db_schema never goes as a startup parameter.

Session state does not survive a transaction. In transaction mode a client owns a server connection from BEGIN to COMMIT and not a moment longer. A plain SET outside a transaction lands on whichever server connection was free and is visible to the next client handed that connection — the last row of the lab: client A ran SET search_path TO leaked, client B read search_path = 'leaked'. The only per-connection state that behaves is state scoped to the transaction (SET LOCAL) or set on the server for the role or database. That is why db_schema is applied as SET LOCAL search_path at the start of every transaction, and why the alternatives are ALTER ROLE … SET, ALTER DATABASE … SET, track_extra_parameters, or schema-qualified metadata — never a SET in a connect listener.

With the statement caches at 0, any PgBouncer works. With statement_cache_size=0 (the default) asyncpg sends unnamed statements and parses them again on every execution, so no statement depends on what a server connection remembers. statement_cache_size > 0 names statements and reuses them in later transactions, which may land on a different server connection. That needs PgBouncer to track prepared statements itself: 1.21 or later with max_prepared_statements > 0. The setting was added in 1.21 at 0 (off) and has defaulted to 200 since 1.24 — measured, a stock 1.21.0, 1.22.1 and 1.23.1 report max_prepared_statements=0, a stock 1.24.1 and 1.25.2 report 200. Without tracking, every cache combination with statement_cache_size > 0 fails with prepared statement "__asyncpg_stmt_…" does not exist. The connection class plays no part: AsyncCConnection and a plain asyncpg.Connection passed and failed in exactly the same runs (issue #60). On a tracking PgBouncer you may raise the caches to win the parse round-trip back.

Pre-ping costs a PgBouncer transaction, and moves PgBouncer's queue out of pool_timeout. pre_ping=True (the default) pings every connection the pool reuses, and SQLAlchemy runs that ping on asyncpg as its own BEGIN / ; / ROLLBACK. Through a transaction-mode pooler that is a separate server assignment before the request's own transaction: ten requests cost twenty PgBouncer transactions (measured, issue #58). And when all of PgBouncer's server connections are busy, the request waits in PgBouncer's queue inside the ping, that is inside pool.connect(): the wait is recorded as connection_checkout_wait_seconds, pool_timeout does not bound it (a pool with timeout=0.2 recorded a 1.0 s checkout wait and no timeout), and only PgBouncer's query_wait_timeout (120 s by default) ends it. Behind a transaction-mode PgBouncer, pre_ping=False is usually the better trade. What the ping buys back is a connection closed while idle: PgBouncer closes a client connection after client_idle_timeout, and without the ping the next request on it fails (InterfaceError: connection is closed) before SQLAlchemy discards it. Set recycle below client_idle_timeout, and below any load balancer's idle timeout, and the pool retires the connection first — measured with client_idle_timeout=2: pre_ping=False, recycle=3600 failed the request after 3 s idle, pre_ping=False, recycle=1 did not.

# Behind PgBouncer in transaction mode
config = BasePostgresConfig(
    connection=ConnectionSettings(host="pgbouncer", port=6432, ...),
    application_name="my-service",
    db_schema="app",   # SET LOCAL at the start of every transaction: arrives
    # jit stays None   # nothing sent: the connection is accepted
    pool=PoolSettings(
        pre_ping=False,  # no extra PgBouncer transaction per checkout
        recycle=300,     # below PgBouncer's client_idle_timeout and any LB idle timeout
    ),
)
manager = create_async_session_manager(config)

What not to do: jit="off" "for pgbouncer" (it is the one setting that makes the connection impossible), extra_server_settings={"search_path": ...} (rejected, or dropped), and a SET search_path in an on_engine_created connect listener (leaks between clients).

Configuration Best Practices¶

1. Use Environment Variables¶

# ✅ Load from environment
settings = Settings()  # Reads from env vars

# ❌ Hardcode secrets
settings = Settings(
    postgres=BasePostgresConfig(
        connection=ConnectionSettings(
            password=SecretStr("hardcoded-secret")  # ❌ Bad
        )
    )
)

2. Separate Configs by Environment¶

# config/development.py
POSTGRES_POOL__SIZE=5
POSTGRES_POOL__MAX_OVERFLOW=10
POSTGRES_QUERY__ECHO=true

# config/production.py
POSTGRES_POOL__SIZE=20
POSTGRES_POOL__MAX_OVERFLOW=30
POSTGRES_QUERY__ECHO=false

3. Use Masked DSN in Logs¶

# ✅ Mask password
logger.info(f"Connecting to: {settings.postgres.to_dsn(mask_password=True)}")
# Connecting to: postgresql+asyncpg://user:**********@localhost:5432/db

# ❌ Expose password
logger.info(f"Connecting to: {settings.postgres.to_dsn()}")
# Connecting to: postgresql+asyncpg://user:secret123@localhost:5432/db

4. Adjust Pool Size for Workload¶

# High-concurrency API
PoolSettings(size=50, max_overflow=50)

# Background workers
PoolSettings(size=5, max_overflow=10)

# Batch processing
PoolSettings(size=2, max_overflow=0)

5. Behind PgBouncer, Send Nothing It Will Not Carry¶

The defaults already do the right thing — statement caches at 0, jit unset, db_schema applied per transaction. Do not put jit, search_path or any other untracked parameter into extra_server_settings. Consider pre_ping=False with a recycle below PgBouncer's client_idle_timeout; see PgBouncer.

6. Enable Metrics in Production¶

BasePostgresConfig(
    metrics_enabled=True,  # Track pool health
    # ...
)

The series to watch is postgres_db_connection_checkout_wait_seconds — how long callers wait for a connection. It climbs before anything fails, and when the wait in the pool's own queue reaches the timeout above, callers start getting sqlalchemy.exc.TimeoutError and postgres_db_connection_timeouts_total starts moving. Behind PgBouncer with pre_ping=True the wait also includes PgBouncer's queue, which timeout does not bound (see PgBouncer). A rising postgres_db_connection_held_duration_seconds with a flat wait is a slower database, not a small pool; a rising wait with a flat held duration is a pool that needs more connections. See Advanced → Prometheus Metrics for the full list.

Next Steps¶

• Advanced Usage — Unit of Work, metrics, telemetry
• Quick Start — Complete working example
• API Reference — Protocol definitions