Service Layer¶

from sqlspec.service import SQLSpecAsyncService, SQLSpecSyncService

Slotted, high-throughput service base classes wrapping SyncDriverAdapterBase and AsyncDriverAdapterBase sessions for clean domain repository and service patterns across Litestar, FastAPI, Starlette, Flask, Sanic, and Dishka. See the Integrations Overview for wiring examples.

Service Base Classes¶

SQLSpecSyncService ¶

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:

Name	Type	Description	Default
`session`	`SyncDriverT \| None`	The caller-owned driver session, mutually exclusive with config.	`None`
`config`	`SyncDatabaseConfig[Any, Any, SyncDriverT] \| NoPoolSyncConfig[Any, SyncDriverT] \| None`	Database configuration used to acquire sessions per helper call.	`None`
`loader`	`SQLFileLoader \| None`	Optional SQL file loader to expose without resolving named queries.	`None`

provide_session ¶

provide_session(session=None)

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

Parameters:

Name	Type	Description	Default
`session`	`SyncDriverT \| None`	Caller-owned override, which this context does not close.	`None`

Returns:

Type	Description
`AbstractContextManager[SyncDriverT]`	A context yielding the driver and releasing only an acquired session.

paginate ¶

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

Execute offset or cursor pagination according to the supplied filter.

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`schema_type`	`type[SchemaT] \| None`	The schema type to map results to.	`None`
`count_with_window`	`bool`	Whether to use COUNT(*) OVER() for offset totals; incompatible with cursor pagination.	`False`
`session`	`SyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`OffsetPagination[SchemaT] \| OffsetPagination[dict[str, Any]] \| CursorPagination[SchemaT] \| CursorPagination[dict[str, Any]]`	An OffsetPagination with a total, or CursorPagination with page tokens.

get_one ¶

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

Fetch one row or raise :class:~sqlspec.exceptions.NotFoundError.

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

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`schema_type`	`type[SchemaT] \| None`	The schema type to map the row to.	`None`
`error_message`	`str \| None`	Optional message for the raised :class:`NotFoundError`.	`None`
`session`	`SyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`SchemaT \| dict[str, Any]`	The single matched row, mapped to `schema_type` when provided.

Raises:

Type	Description
`NotFoundError`	If the query returns zero rows.

exists ¶

exists(statement, /, *parameters, session=None, **kwargs)

Check if any rows exist for the given query.

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`session`	`SyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`bool`	True if at least one row exists, False otherwise.

SQLSpecAsyncService ¶

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:

Name	Type	Description	Default
`session`	`AsyncDriverT \| None`	The caller-owned driver session, mutually exclusive with config.	`None`
`config`	`AsyncDatabaseConfig[Any, Any, AsyncDriverT] \| NoPoolAsyncConfig[Any, AsyncDriverT] \| None`	Database configuration used to acquire sessions per helper call.	`None`
`loader`	`SQLFileLoader \| None`	Optional SQL file loader to expose without resolving named queries.	`None`

provide_session ¶

provide_session(session=None)

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

Parameters:

Name	Type	Description	Default
`session`	`AsyncDriverT \| None`	Caller-owned override, which this context does not close.	`None`

Returns:

Type	Description
`AbstractAsyncContextManager[AsyncDriverT]`	A context yielding the driver and releasing only an acquired session.

paginate ¶

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

Execute offset or cursor pagination according to the supplied filter.

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`schema_type`	`type[SchemaT] \| None`	The schema type to map results to.	`None`
`count_with_window`	`bool`	Whether to use COUNT(*) OVER() for offset totals; incompatible with cursor pagination.	`False`
`session`	`AsyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`OffsetPagination[SchemaT] \| OffsetPagination[dict[str, Any]] \| CursorPagination[SchemaT] \| CursorPagination[dict[str, Any]]`	An OffsetPagination with a total, or CursorPagination with page tokens.

get_one ¶

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

Fetch one row or raise :class:~sqlspec.exceptions.NotFoundError.

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

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`schema_type`	`type[SchemaT] \| None`	The schema type to map the row to.	`None`
`error_message`	`str \| None`	Optional message for the raised :class:`NotFoundError`.	`None`
`session`	`AsyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`SchemaT \| dict[str, Any]`	The single matched row, mapped to `schema_type` when provided.

Raises:

Type	Description
`NotFoundError`	If the query returns zero rows.

exists ¶

exists(statement, /, *parameters, session=None, **kwargs)

Check if any rows exist for the given query.

Parameters:

Name	Type	Description	Default
`statement`	`Statement \| QueryBuilder`	The SQL statement or QueryBuilder instance.	required
`*parameters`	`StatementParameters \| StatementFilter`	Statement parameters or filters.	`()`
`session`	`AsyncDriverT \| None`	Caller-owned driver override; no new session is acquired.	`None`
`**kwargs`	`Any`	Additional keyword arguments for the driver.	`{}`

Returns:

Type	Description
`bool`	True if at least one row exists, False otherwise.