Base¶

The sqlspec.base module defines the SQLSpec registry that owns configuration, connection lifecycles, and session creation.

Example¶

sqlspec registry¶

from sqlspec import SQLSpec

spec = SQLSpec()
spec.add_named_sql("health_check", "select 1 as ok")
query = spec.get_sql("health_check")

Core Responsibilities¶

Register database configurations.
Provide sync and async session context managers.
Manage connection pool startup and shutdown.
Track configs by bind key for multi-database setups.

Session Management¶

provide_session yields a session bound to a specific config.
Sync and async sessions share the same registry API.
Close pools explicitly with close_all_pools when needed.

API Reference¶

class sqlspec.base.SQLSpec[source]¶

Bases: object

Configuration manager and registry for database connections and pools.

__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:: Self

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:: Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]
Returns:: The same configuration instance (it IS the handle).

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

Access the registry of database configurations.

Returns:: Dictionary mapping config instance IDs to config instances.

event_channel(config)[source]¶

Create an event channel for the provided configuration.

Returns SyncEventChannel for sync configs, AsyncEventChannel for async configs.

Parameters:: config¶ (Union[type[Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]], SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any]]) – A registered database configuration instance or type.
Return type:: SyncEventChannel | AsyncEventChannel
Returns:: The appropriate event channel type for the configuration.

telemetry_snapshot()[source]¶

Return aggregated diagnostics across all registered configurations.

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

get_connection(config)[source]¶

Get a database connection for the specified configuration.

Parameters:: config¶ (Union[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 instance.
Return type:: Union[TypeVar(ConnectionT), Awaitable[TypeVar(ConnectionT)]]
Returns:: A database connection or an awaitable yielding a connection.

get_session(config)[source]¶

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

Parameters:: config¶ (Union[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 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(config, *args, **kwargs)[source]¶

Create and provide a database connection from the specified configuration.

Parameters:

config¶ (Union[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 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(config, *args, **kwargs)[source]¶

Create and provide a database session from the specified configuration.

Parameters:

config¶ (Union[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 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(config)[source]¶

Get the connection pool for the specified configuration.

Parameters:: config¶ (Union[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 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(config)[source]¶

Close the connection pool for the specified configuration.

Parameters:: config¶ (Union[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 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 statement/builder cache.
fragment_cache_size¶ (int | None) – Size of the expression/parameter/file cache.
optimized_cache_size¶ (int | None) – Size of the optimized expression cache.
sql_cache_enabled¶ (bool | None) – Enable/disable statement and builder cache.
fragment_cache_enabled¶ (bool | None) – Enable/disable expression/parameter/file cache.
optimized_cache_enabled¶ (bool | None) – Enable/disable optimized expression 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.

Base¶

Example¶

Core Responsibilities¶

Session Management¶

API Reference¶

See Also¶