Base¶

The sqlspec.base module contains the main entry point for SQLSpec: the SQLSpec class that manages database configurations and provides session context managers.

SQLSpec Registry¶

class sqlspec.base.SQLSpec[source]¶

Bases: object

Configuration manager and registry for database connections and pools.

The main SQLSpec registry that manages database configurations and provides sessions.

Key responsibilities:

Register database configurations
Manage connection pool lifecycles
Provide database sessions via context managers
Support multiple databases simultaneously

Example:

from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig

sql = SQLSpec()
db = sql.add_config(
    AsyncpgConfig(
        pool_config={"host": "localhost", "database": "mydb"}
    )
)

async with sql.provide_session(db) as session:
    result = await session.execute("SELECT * FROM users")

__init__(*, loader=None, observability_config=None)[source]¶

async close_all_pools()[source]¶

Explicitly close all connection pools (async and sync).

This method should be called before application shutdown for proper cleanup.

Return type:: None

async __aenter__()[source]¶

Async context manager entry.

Return type:: SQLSpec

async __aexit__(_exc_type, _exc_val, _exc_tb)[source]¶

Async context manager exit with automatic cleanup.

Return type:: None

add_config(config)[source]¶

Add a configuration instance to the registry.

Parameters:: config¶ (Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]) – The configuration instance to add.
Return type:: type[Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]]
Returns:: The type of the added configuration for use as a registry key.

get_config(name)[source]¶

Retrieve a configuration instance by its type or a key.

Parameters:: name¶ – The type of the configuration or a key associated with it.
Returns:: The configuration instance.
Raises:: KeyError – If the configuration is not found.

property configs: dict[type, DatabaseConfigProtocol[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')]]¶

Access the registry of database configurations.

Returns:: Dictionary mapping config types to config instances.

telemetry_snapshot()[source]¶

Return aggregated diagnostics across all registered configurations.

Return type:: dict[str, typing.Any]

get_connection(name)[source]¶

Get a database connection for the specified configuration.

Parameters:: name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
Return type:: Union[TypeVar(ConnectionT), Awaitable[TypeVar(ConnectionT)]]
Returns:: A database connection or an awaitable yielding a connection.

get_session(name)[source]¶

Get a database session (driver adapter) for the specified configuration.

Parameters:: name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
Return type:: Union[TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase), Awaitable[TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]
Returns:: A driver adapter instance or an awaitable yielding one.

provide_connection(name, *args, **kwargs)[source]¶

Create and provide a database connection from the specified configuration.

Parameters:

name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
*args¶ (Any) – Positional arguments to pass to the config’s provide_connection.
**kwargs¶ (Any) – Keyword arguments to pass to the config’s provide_connection.

Return type:

AbstractContextManager[TypeVar(ConnectionT)] | AbstractAsyncContextManager[TypeVar(ConnectionT)]

Returns:

A sync or async context manager yielding a connection.

provide_session(name, *args, **kwargs)[source]¶

Create and provide a database session from the specified configuration.

Parameters:

name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
*args¶ (Any) – Positional arguments to pass to the config’s provide_session.
**kwargs¶ (Any) – Keyword arguments to pass to the config’s provide_session.

Return type:

AbstractContextManager[TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)] | AbstractAsyncContextManager[TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]

Returns:

A sync or async context manager yielding a driver adapter instance.

get_pool(name)[source]¶

Get the connection pool for the specified configuration.

Parameters:: name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
Return type:: type[TypeVar(PoolT)] | Awaitable[type[TypeVar(PoolT)]] | None
Returns:: The connection pool, an awaitable yielding the pool, or None if not supported.

close_pool(name)[source]¶

Close the connection pool for the specified configuration.

Parameters:: name¶ (Union[type[NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], type[AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]], NoPoolSyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], SyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], NoPoolAsyncConfig[TypeVar(ConnectionT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)], AsyncDatabaseConfig[TypeVar(ConnectionT), TypeVar(PoolT), TypeVar(DriverT, bound= SyncDriverAdapterBase | AsyncDriverAdapterBase)]]) – The configuration name or instance.
Return type:: Awaitable[None] | None
Returns:: None, or an awaitable if closing an async pool.

static get_cache_config()[source]¶

Get the current global cache configuration.

Return type:: CacheConfig
Returns:: The current cache configuration.

static update_cache_config(config)[source]¶

Update the global cache configuration.

Parameters:: config¶ (CacheConfig) – The new cache configuration to apply.
Return type:: None

static get_cache_stats()[source]¶

Get current cache statistics.

Return type:: dict[str, typing.Any]
Returns:: Cache statistics object with detailed metrics.

static reset_cache_stats()[source]¶

Reset all cache statistics to zero.

Return type:: None

static log_cache_stats()[source]¶

Log current cache statistics using the configured logger.

Return type:: None

static configure_cache(*, sql_cache_size=None, fragment_cache_size=None, optimized_cache_size=None, sql_cache_enabled=None, fragment_cache_enabled=None, optimized_cache_enabled=None)[source]¶

Update cache configuration with partial values.

Parameters:

sql_cache_size¶ (int | None) – Size of the SQL statement cache.
fragment_cache_size¶ (int | None) – Size of the AST fragment cache.
optimized_cache_size¶ (int | None) – Size of the optimized expression cache.
sql_cache_enabled¶ (bool | None) – Enable/disable SQL cache.
fragment_cache_enabled¶ (bool | None) – Enable/disable fragment cache.
optimized_cache_enabled¶ (bool | None) – Enable/disable optimized cache.

Return type:

None

load_sql_files(*paths)[source]¶

Load SQL files from paths or directories.

Parameters:: *paths¶ (str | Path) – One or more file paths or directory paths to load.
Return type:: None

add_named_sql(name, sql, dialect=None)[source]¶

Add a named SQL query directly.

Parameters:

name¶ (str) – Name for the SQL query.
sql¶ (str) – Raw SQL content.
dialect¶ (str | None) – Optional dialect for the SQL statement.

Return type:

None

get_sql(name)[source]¶

Get a SQL object by name.

Parameters:: name¶ (str) – Name of the statement from SQL file comments. Hyphens in names are converted to underscores.
Return type:: SQL
Returns:: SQL object ready for execution.

list_sql_queries()[source]¶

List all available query names.

Return type:: list[str]
Returns:: Sorted list of query names.

has_sql_query(name)[source]¶

Check if a SQL query exists.

Parameters:: name¶ (str) – Query name to check.
Return type:: bool
Returns:: True if the query exists in the loader.

clear_sql_cache()[source]¶

Clear the SQL file cache.

Return type:: None

reload_sql_files()[source]¶

Reload all SQL files.

Note

This clears the cache and requires calling load_sql_files again.

Return type:: None

get_sql_files()[source]¶

Get list of loaded SQL files.

Return type:: list[str]
Returns:: Sorted list of file paths.

Configuration Types¶

All database adapter configurations inherit from base protocol classes defined in sqlspec.config.

Connection Pooling¶

Connection pooling is configured via adapter-specific TypedDicts passed to the pool_config parameter.

Session Management¶

Session Protocols¶

Sessions are provided by driver adapter classes: SyncDriverAdapterBase and AsyncDriverAdapterBase.

Context Managers¶

Sessions are provided via context managers that ensure proper resource cleanup:

# Sync session
with sql.provide_session(config) as session:
    result = session.execute("SELECT 1")

# Async session
async with sql.provide_session(config) as session:
    result = await session.execute("SELECT 1")

Lifecycle Management¶

Connection Pool Lifecycle¶

SQLSpec automatically manages connection pool lifecycles:

Startup - Pools created when first session requested
Runtime - Pools shared across sessions
Shutdown - Pools closed when SQLSpec instance destroyed

Manual lifecycle control:

sql = SQLSpec()
db = sql.add_config(AsyncpgConfig(pool_config={...}))

# Startup pools explicitly
# Pools created lazily on first use

# ... application runs ...

# Shutdown pools explicitly
await sql.close_all_pools()

Configuration Registry¶

Multiple Database Support¶

SQLSpec supports multiple databases simultaneously:

sql = SQLSpec()

# Add PostgreSQL
pg_db = sql.add_config(AsyncpgConfig(...))

# Add SQLite
sqlite_db = sql.add_config(SqliteConfig(...))

# Get sessions for each
async with sql.provide_session(pg_db) as pg_session:
    users = await pg_session.select("SELECT * FROM users")

async with sql.provide_session(sqlite_db) as sqlite_session:
    cache = await sqlite_session.select("SELECT * FROM cache")

Configuration Lookup¶

Configurations can be retrieved by type or instance:

# By config type
async with sql.provide_session(AsyncpgConfig) as session:
    ...

# By config instance
async with sql.provide_session(config) as session:
    ...

# Get config by type
config = sql.get_config(AsyncpgConfig)

Base¶

SQLSpec Registry¶

Configuration Types¶

Connection Pooling¶

Session Management¶

Session Protocols¶

Context Managers¶

Lifecycle Management¶

Connection Pool Lifecycle¶

Configuration Registry¶

Multiple Database Support¶

Configuration Lookup¶

See Also¶