Dependency Injection

The SQLSpec plugin integrates with Litestar's dependency injection system. By default, it provides a session under the key db_session. You can customize this key or register multiple databases with distinct keys.

Default Injection

When you add SQLSpecPlugin to your app, handlers can request db_session to receive a session scoped to the request:

from sqlspec.adapters.aiosqlite import AiosqliteDriver

@get("/users")
async def list_users(db_session: AiosqliteDriver) -> list[User]:
    result = await db_session.execute("SELECT * FROM users")
    return result.all(schema_type=User)

Note

Use the driver type that matches your config. For example, SqliteDriver for SqliteConfig, AiosqliteDriver for AiosqliteConfig, or AsyncpgDriver for AsyncpgConfig.

Custom Keys

Use extension_config to customize the dependency injection keys for each database. Set session_key, connection_key, and pool_key to unique values when using multiple databases.

dependency keys

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

sqlspec = SQLSpec()
sqlspec.add_config(
    SqliteConfig(
        connection_config={"database": ":memory:"}, extension_config={"litestar": {"session_key": "analytics"}}
    ),
    name="analytics",
)
sqlspec.add_config(
    SqliteConfig(
        connection_config={"database": ":memory:"}, extension_config={"litestar": {"session_key": "primary"}}
    ),
    name="primary",
)

Multiple Databases

Configure each database with unique keys in its extension_config, then use a single plugin. You can mix async and sync adapters - for example, an async PostgreSQL primary database with a sync DuckDB for ETL operations:

from litestar import Litestar, get
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig, AsyncpgDriver
from sqlspec.adapters.duckdb import DuckDBConfig, DuckDBDriver
from sqlspec.extensions.litestar import SQLSpecPlugin

sqlspec = SQLSpec()

# Primary async PostgreSQL database
sqlspec.add_config(
    AsyncpgConfig(
        connection_config={
            "host": "localhost",
            "port": 5432,
            "database": "app",
            "user": "app",
            "password": "secret",
        },
        extension_config={"litestar": {"session_key": "db"}}
    )
)

# ETL sync DuckDB database with custom keys
sqlspec.add_config(
    DuckDBConfig(
        connection_config={"database": "/tmp/etl.db"},
        extension_config={
            "litestar": {
                "session_key": "etl_db",
                "connection_key": "etl_connection",
                "pool_key": "etl_pool",
            }
        }
    )
)

@get("/report")
async def report(db: AsyncpgDriver, etl_db: DuckDBDriver) -> dict:
    # Async query to primary PostgreSQL
    users = await db.select("SELECT * FROM users")
    # Sync query to DuckDB ETL database
    metrics = etl_db.select("SELECT * FROM analytics")
    return {"users": users.all(), "metrics": metrics.all()}

app = Litestar(
    route_handlers=[report],
    plugins=[SQLSpecPlugin(sqlspec=sqlspec)]  # Single plugin handles all configs
)

Config Lookup Outside App Construction

SQLSpecPlugin.get_config() accepts two kinds of identifier, and they become available at different points in the application lifecycle:

• Registry identities - a config instance, its concrete config type, or its bind_key. These exist as soon as the plugin is constructed, so composition roots, CLI commands, and background workers can resolve a config without building a Litestar application first.

• Litestar dependency identities - session_key, connection_key, and pool_key. These name dependency providers generated during plugin registration, so they resolve only after the plugin has been added to a Litestar application.

from litestar import Litestar
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.extensions.litestar import SQLSpecPlugin

sqlspec = SQLSpec()
sqlspec.add_config(
    AsyncpgConfig(
        bind_key="primary",
        connection_config={
            "host": "localhost",
            "port": 5432,
            "database": "app",
            "user": "app",
            "password": "secret",
        },
    )
)

plugin = SQLSpecPlugin(sqlspec=sqlspec)

# Registry identity: resolves before the app exists
primary = plugin.get_config("primary")

app = Litestar(route_handlers=[], plugins=[plugin])

# Dependency identity: resolves only after registration
session_config = plugin.get_config("db_session")

Give each database a distinct bind_key when several share the same config class. A config type resolves only when exactly one configuration uses it; a repeated type raises KeyError listing the candidate bind keys so you can pick the right one. A bind_key takes precedence over a dependency key of the same value in get_config(), while the request-scoped accessors always read strings as dependency keys.

Advanced DuckDB Configuration

DuckDB supports extensions and connection hooks for advanced use cases like attaching external PostgreSQL databases. Use driver_features to configure extensions and on_connection_create for custom connection initialization:

from typing import Any
from sqlspec import SQLSpec
from sqlspec.adapters.duckdb import DuckDBConfig, DuckDBExtensionConfig

def on_connection_create(connection: Any) -> None:
    """Configure DuckDB connection with PostgreSQL attachment."""
    # Load postgres extension and attach external database
    connection.execute("LOAD postgres")
    connection.execute(
        "ATTACH 'dbname=app user=app password=secret host=localhost' "
        "AS pg (TYPE POSTGRES, SCHEMA 'public')"
    )

sqlspec = SQLSpec()
sqlspec.add_config(
    DuckDBConfig(
        connection_config={
            "database": "/tmp/analytics.db",
            "temp_directory": "/tmp",
        },
        driver_features={
            "extensions": [
                DuckDBExtensionConfig(name="postgres"),
                DuckDBExtensionConfig(name="encodings"),
            ],
            "on_connection_create": on_connection_create,
        },
        extension_config={
            "litestar": {
                "session_key": "etl_db",
                "connection_key": "etl_connection",
            }
        }
    )
)

This pattern enables querying PostgreSQL tables directly from DuckDB SQL:

@get("/sync-users")
def sync_users(etl_db: DuckDBDriver) -> dict:
    # Query PostgreSQL via DuckDB's postgres extension
    result = etl_db.execute(
        "INSERT INTO local_users SELECT * FROM pg.users RETURNING *"
    )
    return {"synced": result.rowcount}