Service Layer Pattern#

Build a service from a database config to fetch rows and run transactions. Each query helper opens a short session and releases it before it returns, even when the query raises.

The service can then outlive a single session. You can still pass a session as the first argument when the caller should own its lifetime.

Parameters and filters passed as *parameters are forwarded directly to the driver, which applies filters to the statement and binds parameters automatically.

Base Service#

SQLSpecAsyncService and SQLSpecSyncService live in sqlspec.service. The five web-framework extensions — litestar, fastapi, flask, starlette, and sanic — each re-export the same two objects, so from sqlspec.extensions.litestar import SQLSpecAsyncService gives you the identical class:

from sqlspec.service import SQLSpecAsyncService, SQLSpecSyncService

Pass exactly one of Service(config=cfg) or Service(session). An optional loader=SQLFileLoader() is available through service.loader; the service does not resolve named SQL for you.

Member	Purpose
`paginate(statement, *parameters, schema_type=None, count_with_window=False, session=None)`	Returns a cursor page when passed a `CursorFilter`. Otherwise returns an offset page with a total count.
`get_one(statement, *parameters, schema_type=None, error_message=None, session=None)`	Returns a row or raises `NotFoundError` when no row matches.
`exists(statement, *parameters, session=None)`	Returns `True` when the query matches at least one row.
`begin_transaction()`	Holds one session across helpers, commits on success, and rolls back on error. Nested blocks run in savepoints.
`provide_session(session=None)`	Yields a session for explicit reuse. Acquired sessions are released on exit.
`begin(session=None)` / `commit(session=None)` / `rollback(session=None)`	Control an available session; they never acquire a disposable session.
`session` / `driver`	The caller's session or the active transaction's driver.
`config`	The database configuration the service was built from, or `None` for session-built services.

Subclass whichever matches your driver and add your own query methods:

Async

from sqlspec.adapters.asyncpg import AsyncpgDriver
from sqlspec.service import SQLSpecAsyncService


class UserService(SQLSpecAsyncService[AsyncpgDriver]):
    """Async service for the users table."""

Sync

from sqlspec.adapters.sqlite import SqliteDriver
from sqlspec.service import SQLSpecSyncService


class UserService(SQLSpecSyncService[SqliteDriver]):
    """Sync service for the users table."""

See Service for the full signatures.

Use a Config for Short Sessions#

The examples below use a local SQLite file. Install sqlspec for the sync example or sqlspec[aiosqlite] for the async example. tmp_path is a pytest temporary directory; use your application's database path outside a test.

Bare helpers each acquire their own session. To share one session without starting a transaction, use provide_session() and pass the yielded driver as session=. Merely entering provide_session() does not bind later helpers to that session.

Sync

from sqlspec import SQLFileLoader, sql
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.service import SQLSpecSyncService

config = SqliteConfig(connection_config={"database": str(tmp_path / "users.sqlite")})
service = SQLSpecSyncService(config=config, loader=SQLFileLoader())
users = sql.select("name").from_("users")

try:
    with service.begin_transaction() as session:
        session.execute("CREATE TABLE users (name TEXT)")
        session.execute("INSERT INTO users VALUES (:name)", name="Ada")

    # Each bare helper acquires and releases its own session.
    assert service.get_one(users) == {"name": "Ada"}
    assert service.exists(users)
    assert service.paginate(users).total == 1

    # Explicit reuse borrows this session without starting a transaction.
    with service.provide_session() as session:
        assert service.exists(users, session=session)
        assert service.get_one(users, session=session) == {"name": "Ada"}

    # Both helpers see the same uncommitted transaction.
    with service.begin_transaction() as session:
        session.execute("INSERT INTO users VALUES (:name)", name="Grace")
        assert service.exists(sql.select("name").from_("users").where_eq("name", "Grace"))
        assert service.paginate(users).total == 2

    try:
        with service.begin_transaction() as session:
            session.execute("INSERT INTO users VALUES (:name)", name="Rolled back")
            msg = "Cancel this change"
            raise ValueError(msg)  # noqa: TRY301 - Demonstrate an application error rolling back the block.
    except ValueError:
        pass

    assert service.paginate(users).total == 2
finally:
    config.close_pool()

Async

from sqlspec import SQLFileLoader, sql
from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.service import SQLSpecAsyncService

config = AiosqliteConfig(connection_config={"database": str(tmp_path / "users.sqlite")})
service = SQLSpecAsyncService(config=config, loader=SQLFileLoader())
users = sql.select("name").from_("users")

try:
    async with service.begin_transaction() as session:
        await session.execute("CREATE TABLE users (name TEXT)")
        await session.execute("INSERT INTO users VALUES (:name)", name="Ada")

    assert await service.get_one(users) == {"name": "Ada"}
    assert await service.exists(users)
    assert (await service.paginate(users)).total == 1

    async with service.provide_session() as session:
        assert await service.exists(users, session=session)
        assert await service.get_one(users, session=session) == {"name": "Ada"}

    async with service.begin_transaction() as session:
        await session.execute("INSERT INTO users VALUES (:name)", name="Grace")
        assert await service.exists(sql.select("name").from_("users").where_eq("name", "Grace"))
        assert (await service.paginate(users)).total == 2

    try:
        async with service.begin_transaction() as session:
            await session.execute("INSERT INTO users VALUES (:name)", name="Rolled back")
            msg = "Cancel this change"
            raise ValueError(msg)  # noqa: TRY301 - Demonstrate an application error rolling back the block.
    except ValueError:
        pass

    assert (await service.paginate(users)).total == 2
finally:
    await config.close_pool()

Both examples leave Ada and Grace in the table. The last transaction rolls back its insert when the block raises. Query helpers do not add commits between calls; begin_transaction() commits once when its block succeeds.

Offset and Cursor Pages#

Use the same paginate() method for either page type. The filter selects the mode. Offset pages include a total count; cursor pages include next and previous tokens and do not run a count query.

from sqlspec.core import CursorFilter, LimitOffsetFilter

query = "SELECT id, name FROM users ORDER BY id"

# Offset page with a total count.
offset_page = await service.paginate(
    query, LimitOffsetFilter(limit=20, offset=40)
)

# Cursor page with next and previous tokens.
cursor_page = await service.paginate(
    query, CursorFilter("id", limit=20, cursor=cursor, secret=signing_secret)
)

Here service is a SQLSpecAsyncService. Pass the request's token as cursor, or None for the first page. Keep signing_secret in server configuration. Pass schema_type=User to either call to return typed items. Sync services expose the same method without await.

With no pagination filter, paginate() keeps its offset-page behavior. Do not mix CursorFilter and LimitOffsetFilter in one call. The count_with_window=True option applies only to offset pages and raises an error with a cursor filter.

For an explicit mode, use paginate_limit_offset() or paginate_cursor(). These methods keep a precise return type even when filters come from a dynamic list:

offset_page = await service.paginate_limit_offset(
    query, *offset_filters, schema_type=User
)  # OffsetPagination[User]

cursor_page = await service.paginate_cursor(
    query, *cursor_filters, schema_type=User
)  # CursorPagination[User]

paginate_cursor() requires one CursorFilter. paginate_limit_offset() rejects cursor filters and preserves default offset behavior when no pagination filter is supplied. Both methods validate the mode before executing the query.

The runnable Cursor Pagination example shows both modes and a return trip to the previous cursor page. That guide also covers key shorthand, NULL values, and token handling.

One Route for Both Page Types#

A Litestar handler can return either page type. User can be a msgspec.Struct; no DTO is needed for this response:

from litestar import get
from litestar.di import NamedDependency
from litestar.params import SkipValidation
from msgspec import Struct

from sqlspec.adapters.asyncpg import AsyncpgDriver
from sqlspec.core import FilterTypes, Pagination
from sqlspec.service import SQLSpecAsyncService


class User(Struct):
    id: int
    name: str

@get("/users")
async def list_users(
    service: NamedDependency[SQLSpecAsyncService[AsyncpgDriver]],
    filters: NamedDependency[SkipValidation[list[FilterTypes]]],
) -> Pagination[User]:
    return await service.paginate(
        "SELECT id, name FROM users ORDER BY id", *filters, schema_type=User
    )

Register dependencies for service and filters on the route or app. The filter dependency must choose one pagination mode per request. Litestar encodes either returned page and describes both response shapes using OpenAPI oneOf. Pagination[User] is an alias for CursorPagination[User] | OffsetPagination[User]; it adds no response wrapper. To document only one mode, annotate the handler with CursorPagination[User] or OffsetPagination[User]. OpenAPI uses the handler return annotation, not the service overload selected within its body.

This response encoding does not imply typed union decoding support. msgspec.json.Decoder(CursorPagination[User] | OffsetPagination[User]) cannot decode the union of these two dataclasses. If a client needs that typed union, define tagged msgspec.Struct response envelopes in your application. See msgspec union restrictions and tagged unions.

Session Ownership#

A passed session= wins over the active transaction. The service borrows it; the caller decides when to commit and release it. Without an override, helpers use the active transaction, then the constructor session, or acquire a new session from the config.

Outside a config service's transaction, session and driver raise ImproperConfigurationError. Manual begin(), commit(), and rollback() also require an available session. Use begin_transaction() or pass a driver explicitly with session= for manual control.

Tasks and threads can use the same service, within the adapter's concurrency limits. A child task of a config service cannot implicitly reuse its parent's transaction, even after that transaction exits. Start independent work outside the parent's transaction context, or pass a session whose use you control.

Nested Transaction Blocks#

A begin_transaction() block entered inside another block on the same service runs in a savepoint on the outer session; no second session is acquired. When the inner block raises, only its work is rolled back and the exception propagates, so the outer block can catch it and keep using the session. This lets an insert that may hit a unique constraint run without aborting the surrounding transaction:

from sqlspec.exceptions import UniqueViolationError

async with service.begin_transaction() as session:
    try:
        async with service.begin_transaction():
            await session.execute("INSERT INTO users (email) VALUES (:email)", email=email)
    except UniqueViolationError:
        pass
    user = await service.get_one("SELECT id, email FROM users WHERE email = :email", email=email)

The outer block commits everything that was not rolled back; if the outer block raises, work from inner blocks that succeeded is rolled back with it. Nesting works the same way for services built from a session, including a block entered inside the session's own transaction() block. When a service built from a session enters its outermost block while that session already has an open transaction, for example after an earlier statement, the block joins that transaction instead of calling begin(), and exiting the block commits or rolls back all of it, including the earlier work. If the outer commit fails, the block attempts a rollback before raising the commit error. Adapters without savepoint support (DuckDB, BigQuery, Spanner, and ADBC connections to DuckDB, BigQuery, or Snowflake) raise ImproperConfigurationError when a nested block is entered.

A nested block must run in the task or thread that entered the outer block. Entering begin_transaction() from another task or thread while the outer block is active raises ImproperConfigurationError instead of starting a second transaction on the same session. The outer block itself may be entered and exited in different tasks or threads, as happens with framework dependencies that run in a thread pool.

Services expose config, the configuration they were built from, so a service can construct a collaborating service without reaching into private state:

audit = AuditService(config=users.config)

audit is a separate service with its own sessions and transactions. It does not join a transaction that users has open; pass session= to run its helpers on that session instead.

Domain Services with a Caller-Owned Session#

Inherit from the base and use the sql builder or SQL file loader for queries. Filters from Litestar dependencies flow straight through *filters to the driver. These examples receive a caller-owned session, so direct self.driver calls remain available. For a config-built service, perform direct driver operations inside begin_transaction() or through an explicitly provided session.

Async

from uuid import UUID

from pydantic import BaseModel
from sqlspec import sql
from sqlspec.adapters.asyncpg import AsyncpgDriver
from sqlspec.core import OffsetPagination, StatementFilter
from sqlspec.service import SQLSpecAsyncService


class User(BaseModel):
    id: str
    email: str
    name: str


class UserService(SQLSpecAsyncService[AsyncpgDriver]):

    async def list_with_count(self, *filters: StatementFilter) -> OffsetPagination[User]:
        return await self.paginate(
            sql.select("id", "email", "name").from_("users"),
            *filters,
            schema_type=User,
        )

    async def get_user(self, user_id: UUID) -> User:
        return await self.get_one(
            sql.select("id", "email", "name").from_("users").where_eq("id", user_id),
            schema_type=User,
            error_message=f"User {user_id} not found",
        )

    async def create_user(self, email: str, name: str) -> User:
        return await self.driver.select_one(
            sql.insert("users").columns("email", "name").values(email, name).returning("id", "email", "name"),
            schema_type=User,
        )

Sync

from uuid import UUID

from pydantic import BaseModel
from sqlspec import sql
from sqlspec.adapters.sqlite import SqliteDriver
from sqlspec.core import OffsetPagination, StatementFilter
from sqlspec.service import SQLSpecSyncService


class User(BaseModel):
    id: str
    email: str
    name: str


class UserService(SQLSpecSyncService[SqliteDriver]):

    def list_with_count(self, *filters: StatementFilter) -> OffsetPagination[User]:
        return self.paginate(
            sql.select("id", "email", "name").from_("users"),
            *filters,
            schema_type=User,
        )

    def get_user(self, user_id: UUID) -> User:
        return self.get_one(
            sql.select("id", "email", "name").from_("users").where_eq("id", user_id),
            schema_type=User,
            error_message=f"User {user_id} not found",
        )

    def create_user(self, email: str, name: str) -> User:
        return self.driver.select_one(
            sql.insert("users").columns("email", "name").values(email, name).returning("id", "email", "name"),
            schema_type=User,
        )

Using with Litestar#

With create_filter_dependencies, filters are injected from query parameters and forwarded through the service to the driver.

This continues the UserService and User from the previous section, and provides the service itself alongside the filter dependencies:

from litestar import Controller, get
from litestar.di import NamedDependency, Provide
from litestar.params import SkipValidation
from sqlspec.adapters.asyncpg import AsyncpgDriver
from sqlspec.core import FilterTypes, OffsetPagination
from sqlspec.extensions.litestar.providers import create_filter_dependencies


async def provide_users_service(db_session: AsyncpgDriver) -> UserService:
    return UserService(db_session)


class UserController(Controller):
    path = "/api/users"
    dependencies = {
        "users_service": Provide(provide_users_service),
        **create_filter_dependencies(
            {
                "pagination_type": "limit_offset",
                "pagination_size": 20,
                "sort_field": ["created_at", "uploaded_collections", "name"],
                "sort_order": "desc",
                "search": "name,email",
            }
        ),
    }

    @get()
    async def list_users(
        self,
        users_service: UserService,
        filters: SkipValidation[NamedDependency[list[FilterTypes]]],
    ) -> OffsetPagination[User]:
        return await users_service.list_with_count(*filters)

sort_field remains the SQL-facing allowlist. Clients can request ?orderBy=uploadedCollections by default while the service receives an OrderByFilter for uploaded_collections. Existing snake_case values, such as ?orderBy=uploaded_collections, remain valid for compatibility. Set sort_field_camelize=False when an endpoint must accept only raw configured orderBy values.