Service Layer¶

SQLSpecSyncService and SQLSpecAsyncService (sqlspec.service) wrap an adapter configuration (or an already-checked-out driver session) with ergonomic helpers for single-row lookups (get_one), existence checks (exists), offset and cursor pagination (paginate, paginate_limit_offset, paginate_cursor), and atomic write blocks (begin_transaction).

Full Sync and Async Service Example¶

When constructed with config=, a service checks out a short-lived pooled session automatically for standalone read helpers (service.get_one(...), service.exists(...), service.paginate(...)) and opens an explicit commit-or-rollback session inside with service.begin_transaction() as session::

docs/examples/drivers/service_config.py

"""Use SQLSpecSyncService and SQLSpecAsyncService for short helper sessions and explicit transactions."""

import asyncio
import tempfile
from pathlib import Path
from typing import NoReturn

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

__all__ = ("main", "run_async_service_example", "run_sync_service_example")


def run_sync_service_example(db_path: Path) -> int:
    """Execute synchronous service helpers and atomic transaction blocks."""
    config = SqliteConfig(connection_config={"database": str(db_path)})
    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")

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

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

        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")
                _abort_transaction()
        except ValueError:
            pass

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


async def run_async_service_example(db_path: Path) -> int:
    """Execute asynchronous service helpers and atomic transaction blocks."""
    config = AiosqliteConfig(connection_config={"database": str(db_path)})
    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")
                _abort_transaction()
        except ValueError:
            pass

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


async def main() -> tuple[int, int]:
    """Run both synchronous and asynchronous service examples in a temporary directory."""
    with tempfile.TemporaryDirectory() as tmp_dir:
        base_dir = Path(tmp_dir)
        sync_total = run_sync_service_example(base_dir / "sync_users.sqlite")
        async_total = await run_async_service_example(base_dir / "async_users.sqlite")
        return sync_total, async_total


def _abort_transaction() -> NoReturn:
    """Raise an error inside a transaction block to demonstrate automatic rollback."""
    msg = "Cancel this change"
    raise ValueError(msg)


if __name__ == "__main__":
    print(asyncio.run(main()))

Core Service Methods¶

Method	Purpose
`get_one(statement, filters, session=None, *params)`	Fetch a single row (or typed `schema_type` instance), raising `NotFoundError` when absent.
`exists(statement, filters, session=None, *params)`	Return `bool` indicating whether at least one row matches the query.
`paginate(statement, filters, session=None, *params)`	Dispatch automatically to limit/offset or cursor pagination based on attached filters.
`paginate_limit_offset(statement, filter, ...)`	Return an `OffsetPagination` envelope (`items`, `limit`, `offset`, `total`).
`paginate_cursor(statement, filter, ...)`	Return a `CursorPagination` envelope (`items`, `next_cursor`, `previous_cursor`).
`provide_session()`	Context manager yielding a pooled read/write driver session.
`begin_transaction()`	Context manager yielding a session inside an active transaction that commits on exit and rolls back on exception.

Structuring Domain Services¶

Subclass or compose SQLSpecAsyncService (or SQLSpecSyncService) around your domain's .sql file loader so route handlers express domain operations rather than SQL mechanics:

import msgspec

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


class Customer(msgspec.Struct, frozen=True, gc=False):
    id: int
    email: str
    tier: str


class CustomerService:
    def __init__(self, config: AiosqliteConfig, loader: SQLFileLoader) -> None:
        self._db = SQLSpecAsyncService(config=config, loader=loader)

    async def get_by_email(self, email: str) -> Customer:
        query = sql.select("id", "email", "tier").from_("customers").where_eq("email", email)
        row = await self._db.get_one(query)
        return msgspec.convert(row, type=Customer)

    async def list_customers(self, *, limit: int = 25, offset: int = 0) -> OffsetPagination[dict[str, object]]:
        query = sql.select("id", "email", "tier").from_("customers").order_by("id")
        return await self._db.paginate_limit_offset(query, LimitOffsetFilter(limit=limit, offset=offset))

    async def upgrade_tier(self, customer_id: int, new_tier: str) -> None:
        async with self._db.begin_transaction() as session:
            await session.execute("UPDATE customers SET tier = :tier WHERE id = :id", tier=new_tier, id=customer_id)
            await session.execute(
                "INSERT INTO tier_audit (customer_id, new_tier) VALUES (:id, :tier)", id=customer_id, tier=new_tier
            )

Pass session=session to get_one, exists, or paginate (or construct SQLSpecAsyncService(session)) when composing multiple domain services inside a single outer begin_transaction() block so every read and write participates in the same atomic transaction:

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.service import SQLSpecSyncService

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
orders_svc = SQLSpecSyncService(config=config)
ledger_svc = SQLSpecSyncService(config=config)

with orders_svc.begin_transaction() as session:
    session.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, total INTEGER)")
    session.execute("CREATE TABLE ledger (id INTEGER PRIMARY KEY, order_id INTEGER, amount INTEGER)")
    session.execute("INSERT INTO orders (id, total) VALUES (:id, :total)", id=101, total=2500)

    order = orders_svc.get_one("SELECT id, total FROM orders WHERE id = :id", session=session, id=101)
    session.execute(
        "INSERT INTO ledger (order_id, amount) VALUES (:order_id, :amount)", order_id=order["id"], amount=order["total"]
    )
    assert ledger_svc.exists("SELECT 1 FROM ledger WHERE order_id = :id", session=session, id=101)

Recap¶

Construct SQLSpecSyncService(config=...) or SQLSpecAsyncService(config=...) to eliminate repetitive with config.provide_session() boilerplate on single-statement reads.
Group multi-statement domain mutations inside service.begin_transaction() for automatic commit on success and rollback on failure.
Pass session=session when coordinating multiple domain services within one shared transaction.

Service Layer¶

Full Sync and Async Service Example¶

Core Service Methods¶

Structuring Domain Services¶

Sharing an Existing Session Across Multiple Services¶

Recap¶