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
from sqlspec.extensions.litestar import SQLSpecPlugin

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

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",
            }
        }
    )
)

from typing import Any

@get("/report")
async def report(db: AsyncpgDriver, etl_db: DuckDBDriver) -> dict[str, Any]:
    # 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, "metrics": metrics}

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

Bringing Your Own DI

Applications that use another dependency injection container can turn off the plugin's dependency providers and per-request commit and close handling with disable_di and still let the plugin create the pool on startup, store it in application state under pool_key, and close it on shutdown with manage_lifespan. Both settings live in each config's extension_config["litestar"] and apply to that config only; disable_di defaults to False and manage_lifespan defaults to not disable_di.

Settings	Dependencies	Pool lifespan
defaults	registered	managed
disable_di=True	not registered	not managed
disable_di=True, manage_lifespan=True	not registered	managed
manage_lifespan=False	registered	not managed

With disable_di=True and manage_lifespan=True, your own providers open sessions from the pool the plugin started:

from collections.abc import AsyncIterator

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

sqlspec = SQLSpec()
config = sqlspec.add_config(
    AsyncpgConfig(
        connection_config={"dsn": "postgresql://app:secret@localhost:5432/app"},
        extension_config={"litestar": {"disable_di": True, "manage_lifespan": True}},
    )
)

async def provide_session() -> AsyncIterator[AsyncpgDriver]:
    async with config.provide_session() as session:
        yield session

app = Litestar(route_handlers=[], plugins=[SQLSpecPlugin(sqlspec=sqlspec)])

Register provide_session with your container; the pool is open for the lifetime of the application and closed when it shuts down.

Note

SQLSpecPlugin.provide_request_session*() and provide_request_connection*() rely on the plugin's dependency providers and per-request close handling, so they do not work with disable_di=True. Open sessions with the context-managed config.provide_session() instead.

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.

An identifier that matches neither a registry identity nor a dependency key raises KeyError listing every available bind key and dependency key. This holds both before and after the plugin is registered with a Litestar application.

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[str, int]:
    # Query PostgreSQL via DuckDB's postgres extension
    result = etl_db.execute(
        "INSERT INTO local_users SELECT * FROM pg.users RETURNING *"
    )
    return {"synced": result.rowcount}