Service#

Base service classes provide pagination, single-row fetching, existence checks, and transactions. Construct them with a caller-owned driver session or a database config=. Config-built helpers acquire short sessions; begin_transaction() holds one session across calls and releases it on exit.

Each query helper accepts a borrowed session= override. provide_session() yields a driver for explicit reuse. The optional loader= is exposed without resolving named queries. config returns the database configuration a service was built from, or None for a service built from a session, so one service can construct another from the same configuration. See the recipe for ownership and concurrency rules.

begin_transaction() blocks can nest on every service. A nested block runs inside a savepoint on the outer block's session and never acquires a second session. It releases the savepoint when it succeeds and rolls back to the savepoint when it raises, then re-raises, so only the inner work is undone and the outer block can continue and commit. Adapters without savepoint support (DuckDB, BigQuery, Spanner, and ADBC connections to DuckDB, BigQuery, or Snowflake) raise ImproperConfigurationError when a nested block is entered.

The five web-framework extensions — litestar, fastapi, flask, starlette, and sanic — each re-export these two objects, so from sqlspec.extensions.litestar import SQLSpecAsyncService gives the identical class. See Service Layer Pattern for usage.

SQLSpecAsyncService#

class sqlspec.service.SQLSpecAsyncService[source]#

Bases: Generic[AsyncDriverT]

Base class for asynchronous SQLSpec services.

Config-built services acquire and release a short session for each query helper. Session-built services borrow the caller's session without closing it.

Parameters:

session¶ (Optional[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]) -- The caller-owned driver session, mutually exclusive with config.
config¶ (Union[AsyncDatabaseConfig[typing.Any, typing.Any, TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)], NoPoolAsyncConfig[typing.Any, TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)], None]) -- Database configuration used to acquire sessions per helper call.
loader¶ (SQLFileLoader | None) -- Optional SQL file loader to expose without resolving named queries.

__init__(session=None, *, config=None, loader=None)[source]#

property session: AsyncDriverT#

Return the active transaction driver or the constructor session.

Raises:: ImproperConfigurationError -- If a config-built service has no active transaction.

property driver: AsyncDriverT#: Alias for session matching the recipe-doc terminology.

property config: AsyncDatabaseConfig[MyTypeAliasForwardRef('typing.Any'), MyTypeAliasForwardRef('typing.Any'), AsyncDriverT] | NoPoolAsyncConfig[MyTypeAliasForwardRef('typing.Any'), AsyncDriverT] | None#: Return the database configuration this service was built from, or None for session-bound services.

property loader: SQLFileLoader | None#: Return the optional SQL file loader.

provide_session(session=None)[source]#

Borrow an available session or acquire a short config-owned session.

Parameters:: session¶ (Optional[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]) -- Caller-owned override, which this context does not close.
Return type:: AbstractAsyncContextManager[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]
Returns:: A context yielding the driver and releasing only an acquired session.

async paginate(statement, /, *parameters, schema_type=None, count_with_window=False, session=None, **kwargs)[source]#

Execute offset or cursor pagination according to the supplied filter.

Overloads:

self, statement (Statement | QueryBuilder), cursor_filter (CursorFilter), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), cursor_filter (CursorFilter), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → CursorPagination[dict[str, Any]]
self, statement (Statement | QueryBuilder), limit_offset_filter (LimitOffsetFilter), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT]
self, statement (Statement | QueryBuilder), limit_offset_filter (LimitOffsetFilter), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT] | CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]] | CursorPagination[dict[str, Any]]

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
schema_type¶ -- The schema type to map results to.
count_with_window¶ -- Whether to use COUNT(*) OVER() for offset totals; incompatible with cursor pagination.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

An OffsetPagination with a total, or CursorPagination with page tokens.

async paginate_limit_offset(statement, /, *parameters, schema_type=None, count_with_window=False, session=None, **kwargs)[source]#

Execute limit/offset pagination with an explicit result type.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (AsyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]]

Parameters:

statement¶ -- SQL statement or query builder.
*parameters¶ -- Statement parameters and pagination filters.
schema_type¶ -- Schema type for result conversion.
count_with_window¶ -- Use a window function to calculate the total.
session¶ -- Caller-owned driver override.
**kwargs¶ -- Additional driver arguments.

Returns:

The limit/offset pagination result.

Raises:

ImproperConfigurationError -- Filters conflict with the requested pagination mode.

async paginate_cursor(statement, /, *parameters, schema_type=None, session=None, **kwargs)[source]#

Execute cursor pagination with an explicit result type.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), session (AsyncDriverT | None), kwargs (Any) → CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), session (AsyncDriverT | None), kwargs (Any) → CursorPagination[dict[str, Any]]

Parameters:

statement¶ -- SQL statement or query builder.
*parameters¶ -- Statement parameters and pagination filters.
schema_type¶ -- Schema type for result conversion.
session¶ -- Caller-owned driver override.
**kwargs¶ -- Additional driver arguments.

Returns:

The cursor pagination result.

Raises:

ImproperConfigurationError -- Filters conflict with the requested pagination mode.

async get_one(statement, /, *parameters, schema_type=None, error_message=None, session=None, **kwargs)[source]#

Fetch one row or raise NotFoundError.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), error_message (str | None), session (AsyncDriverT | None), kwargs (Any) → SchemaT
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), error_message (str | None), session (AsyncDriverT | None), kwargs (Any) → dict[str, Any]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT] | None), error_message (str | None), session (AsyncDriverT | None), kwargs (Any) → SchemaT | dict[str, Any]

HTTP status mapping is the responsibility of the calling framework integration. The Litestar extension registers a default mapping; other framework integrations do not.

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
schema_type¶ -- The schema type to map the row to.
error_message¶ -- Optional message for the raised NotFoundError.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

The single matched row, mapped to schema_type when provided.

Raises:

NotFoundError -- If the query returns zero rows.

async exists(statement, /, *parameters, session=None, **kwargs)[source]#

Check if any rows exist for the given query.

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

True if at least one row exists, False otherwise.

async begin(*, session=None)[source]#

Begin a database transaction on the underlying session.

Parameters:: session¶ (Optional[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

async commit(*, session=None)[source]#

Commit the current database transaction.

Parameters:: session¶ (Optional[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

async rollback(*, session=None)[source]#

Roll back the current database transaction.

Parameters:: session¶ (Optional[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

begin_transaction()[source]#

Context manager that commits on success and rolls back on error.

Nested blocks run in a savepoint on the outer session.

Return type:: _AsyncBeginTransactionContext[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]
Returns:: The underlying driver session bound to the active transaction.

SQLSpecSyncService#

class sqlspec.service.SQLSpecSyncService[source]#

Bases: Generic[SyncDriverT]

Base class for synchronous SQLSpec services.

Config-built services acquire and release a short session for each query helper. Session-built services borrow the caller's session without closing it.

Parameters:

session¶ (Optional[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]) -- The caller-owned driver session, mutually exclusive with config.
config¶ (Union[SyncDatabaseConfig[typing.Any, typing.Any, TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)], NoPoolSyncConfig[typing.Any, TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)], None]) -- Database configuration used to acquire sessions per helper call.
loader¶ (SQLFileLoader | None) -- Optional SQL file loader to expose without resolving named queries.

__init__(session=None, *, config=None, loader=None)[source]#

property session: SyncDriverT#

Return the active transaction driver or the constructor session.

Raises:: ImproperConfigurationError -- If a config-built service has no active transaction.

property driver: SyncDriverT#: Alias for session matching the recipe-doc terminology.

property config: SyncDatabaseConfig[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any'), SyncDriverT] | NoPoolSyncConfig[TypeAliasForwardRef('typing.Any'), SyncDriverT] | None#: Return the database configuration this service was built from, or None for session-bound services.

property loader: SQLFileLoader | None#: Return the optional SQL file loader.

provide_session(session=None)[source]#

Borrow an available session or acquire a short config-owned session.

Parameters:: session¶ (Optional[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]) -- Caller-owned override, which this context does not close.
Return type:: AbstractContextManager[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]
Returns:: A context yielding the driver and releasing only an acquired session.

paginate(statement, /, *parameters, schema_type=None, count_with_window=False, session=None, **kwargs)[source]#

Execute offset or cursor pagination according to the supplied filter.

Overloads:

self, statement (Statement | QueryBuilder), cursor_filter (CursorFilter), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), cursor_filter (CursorFilter), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → CursorPagination[dict[str, Any]]
self, statement (Statement | QueryBuilder), limit_offset_filter (LimitOffsetFilter), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT]
self, statement (Statement | QueryBuilder), limit_offset_filter (LimitOffsetFilter), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT] | CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]] | CursorPagination[dict[str, Any]]

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
schema_type¶ -- The schema type to map results to.
count_with_window¶ -- Whether to use COUNT(*) OVER() for offset totals; incompatible with cursor pagination.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

An OffsetPagination with a total, or CursorPagination with page tokens.

paginate_limit_offset(statement, /, *parameters, schema_type=None, count_with_window=False, session=None, **kwargs)[source]#

Execute limit/offset pagination with an explicit result type.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), session (SyncDriverT | None), kwargs (Any) → OffsetPagination[dict[str, Any]]

Parameters:

statement¶ -- SQL statement or query builder.
*parameters¶ -- Statement parameters and pagination filters.
schema_type¶ -- Schema type for result conversion.
count_with_window¶ -- Use a window function to calculate the total.
session¶ -- Caller-owned driver override.
**kwargs¶ -- Additional driver arguments.

Returns:

The limit/offset pagination result.

Raises:

ImproperConfigurationError -- Filters conflict with the requested pagination mode.

paginate_cursor(statement, /, *parameters, schema_type=None, session=None, **kwargs)[source]#

Execute cursor pagination with an explicit result type.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), session (SyncDriverT | None), kwargs (Any) → CursorPagination[SchemaT]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), session (SyncDriverT | None), kwargs (Any) → CursorPagination[dict[str, Any]]

Parameters:

statement¶ -- SQL statement or query builder.
*parameters¶ -- Statement parameters and pagination filters.
schema_type¶ -- Schema type for result conversion.
session¶ -- Caller-owned driver override.
**kwargs¶ -- Additional driver arguments.

Returns:

The cursor pagination result.

Raises:

ImproperConfigurationError -- Filters conflict with the requested pagination mode.

get_one(statement, /, *parameters, schema_type=None, error_message=None, session=None, **kwargs)[source]#

Fetch one row or raise NotFoundError.

Overloads:

self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), error_message (str | None), session (SyncDriverT | None), kwargs (Any) → SchemaT
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), error_message (str | None), session (SyncDriverT | None), kwargs (Any) → dict[str, Any]
self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT] | None), error_message (str | None), session (SyncDriverT | None), kwargs (Any) → SchemaT | dict[str, Any]

HTTP status mapping is the responsibility of the calling framework integration. The Litestar extension registers a default mapping; other framework integrations do not.

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
schema_type¶ -- The schema type to map the row to.
error_message¶ -- Optional message for the raised NotFoundError.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

The single matched row, mapped to schema_type when provided.

Raises:

NotFoundError -- If the query returns zero rows.

exists(statement, /, *parameters, session=None, **kwargs)[source]#

Check if any rows exist for the given query.

Parameters:

statement¶ -- The SQL statement or QueryBuilder instance.
*parameters¶ -- Statement parameters or filters.
session¶ -- Caller-owned driver override; no new session is acquired.
**kwargs¶ -- Additional keyword arguments for the driver.

Returns:

True if at least one row exists, False otherwise.

begin(*, session=None)[source]#

Begin a database transaction on the underlying session.

Parameters:: session¶ (Optional[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

commit(*, session=None)[source]#

Commit the current database transaction.

Parameters:: session¶ (Optional[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

rollback(*, session=None)[source]#

Roll back the current database transaction.

Parameters:: session¶ (Optional[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]) -- Caller-owned driver override for manual transaction control.
Return type:: None

begin_transaction()[source]#

Context manager that commits on success and rolls back on error.

Nested blocks run in a savepoint on the outer session.

Return type:: _SyncBeginTransactionContext[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]
Returns:: The underlying driver session bound to the active transaction.