Service

Base service classes that wrap a driver session and add pagination, single-row fetching, existence checks, and transaction helpers.

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

Bases: Generic[AsyncDriverT]

Base class for asynchronous SQLSpec services.

Provides common database operations and pagination support using a driver session.

Parameters:

session (TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)) -- The driver session instance.

__init__(session)

property session: AsyncDriverT

Return the driver session.

property driver: AsyncDriverT

Alias for session matching the recipe-doc terminology.

async paginate(statement, /, *parameters, schema_type=None, count_with_window=False, **kwargs)

Execute a paginated query and return an OffsetPagination container.

Overloads:

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), kwargs (Any) → OffsetPagination[SchemaT]

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), kwargs (Any) → OffsetPagination[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 total count.

• **kwargs -- Additional keyword arguments for the driver.

Returns:

An OffsetPagination instance containing items and total count.

async get_one(statement, /, *parameters, schema_type=None, error_message=None, **kwargs)

Fetch one row or raise NotFoundError.

Overloads:

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

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

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT] | None), error_message (str | 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.

• **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, **kwargs)

Check if any rows exist for the given query.

Parameters:

• statement -- The SQL statement or QueryBuilder instance.

• *parameters -- Statement parameters or filters.

• **kwargs -- Additional keyword arguments for the driver.

Returns:

True if at least one row exists, False otherwise.

async begin()

Begin a database transaction on the underlying session.

Return type:

None

async commit()

Commit the current database transaction.

Return type:

None

async rollback()

Roll back the current database transaction.

Return type:

None

begin_transaction()

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

Return type:

_AsyncBeginTransactionContext[TypeVar(AsyncDriverT, bound= AsyncDriverAdapterBase)]

Returns:

The underlying driver session bound to the active transaction.

SQLSpecSyncService

class sqlspec.service.SQLSpecSyncService

Bases: Generic[SyncDriverT]

Base class for synchronous SQLSpec services.

Provides common database operations and pagination support using a driver session.

Parameters:

session (TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)) -- The driver session instance.

__init__(session)

property session: SyncDriverT

Return the driver session.

property driver: SyncDriverT

Alias for session matching the recipe-doc terminology.

paginate(statement, /, *parameters, schema_type=None, count_with_window=False, **kwargs)

Execute a paginated query and return an OffsetPagination container.

Overloads:

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), count_with_window (bool), kwargs (Any) → OffsetPagination[SchemaT]

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), count_with_window (bool), kwargs (Any) → OffsetPagination[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 total count.

• **kwargs -- Additional keyword arguments for the driver.

Returns:

An OffsetPagination instance containing items and total count.

get_one(statement, /, *parameters, schema_type=None, error_message=None, **kwargs)

Fetch one row or raise NotFoundError.

Overloads:

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

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

• self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT] | None), error_message (str | 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.

• **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, **kwargs)

Check if any rows exist for the given query.

Parameters:

• statement -- The SQL statement or QueryBuilder instance.

• *parameters -- Statement parameters or filters.

• **kwargs -- Additional keyword arguments for the driver.

Returns:

True if at least one row exists, False otherwise.

begin()

Begin a database transaction on the underlying session.

Return type:

None

commit()

Commit the current database transaction.

Return type:

None

rollback()

Roll back the current database transaction.

Return type:

None

begin_transaction()

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

Return type:

_SyncBeginTransactionContext[TypeVar(SyncDriverT, bound= SyncDriverAdapterBase)]

Returns:

The underlying driver session bound to the active transaction.