Skip to content

Database Engine & ORM

ZodiacCore provides a high-performance, async-first database abstraction layer built on top of SQLModel and SQLAlchemy 2.0. It simplifies session management, connection pooling, and standardizes model definitions.

1. Core Concepts

The Database Manager

The DatabaseManager (exposed as the global db instance) is a strict singleton that manages the SQLAlchemy AsyncEngine and async_sessionmaker. It ensures that your process can reuse connection pools for the same named database instead of letting each app/container create its own pool, which is critical for performance and resource management.

The Repository Pattern

We encourage the use of the Repository Pattern via BaseSQLRepository. This decouples your business logic from database-specific code, making your application more maintainable and easier to unit test with mocks.

2. Model Definitions

ZodiacCore provides several mixins and base classes in zodiac_core.db.sql to standardize your database schema.

Standard Base Models

Instead of inheriting from SQLModel directly, we recommend using our pre-configured base models:

Base Model Primary Key Timestamps
IntIDModel id: int (Auto-increment) created_at, updated_at
UUIDModel id: UUID (v4) created_at, updated_at

Example: Using Base Models

from zodiac_core.db.sql import IntIDModel
from sqlmodel import Field

class User(IntIDModel, table=True):
    username: str = Field(unique=True, index=True)
    email: str

Automatic Timestamps

Both IntIDModel and UUIDModel include SQLDateTimeMixin, which provides:

  • created_at: Automatically set on insertion.
  • updated_at: Automatically updated on every save via a SQLAlchemy event listener.

3. Configuration & Lifecycle

You should initialize the database during your application's startup and ensure it shuts down cleanly. Calling db.setup(...) again with the same name is allowed only when the effective configuration is identical; different settings for an existing name raise RuntimeError.

For multi‑app deployments with app.mount(), see the Sub Applications guide. Lifecycle control is now name-aware:

  • await db.shutdown(name="...") disposes only the selected named database.
  • await db.shutdown() disposes all registered databases.

This lets multiple apps, containers, or resources share the global manager while still releasing only the resource they own.

FastAPI Integration

We recommend using the lifespan context manager (FastAPI 0.93+). The legacy on_event("startup") / on_event("shutdown") are deprecated.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from zodiac_core.db import db


@asynccontextmanager
async def lifespan(app: FastAPI):
    db.setup(
        "postgresql+asyncpg://user:pass@localhost/dbname",
        pool_size=20,
        max_overflow=10,
        echo=False,
    )
    await db.create_all()  # Optional: create tables if they don't exist
    yield
    await db.shutdown()


app = FastAPI(lifespan=lifespan)

For a single-app service, await db.shutdown() is still the simplest shutdown path. If you register multiple named databases or share the global db across multiple app lifecycles, prefer await db.shutdown(name="...") for scoped cleanup.

4. Choosing a Session API

The session APIs are intentionally not interchangeable. Choose the API from the call site that owns the session lifecycle:

Call site Use
FastAPI endpoint owns a unit of work on the default database Depends(get_session)
FastAPI endpoint owns a unit of work on a named database A module-level dependency created by session_dependency(name)
Lower layer participates in an endpoint-owned unit of work Accept the concrete AsyncSession from its caller
Service, repository, job, CLI, or startup task owns the unit of work BaseSQLRepository.session() or async with db.session(name)

FastAPI dependencies are not general session APIs

get_session retains its optional name argument for compatibility with ZodiacCore 0.7.0. When FastAPI resolves Depends(get_session), however, the name comes from private server-side wiring and always selects the default database; it is never read from the request.

For new named routes, call session_dependency(name) once at module or router scope and pass the stored callable to Depends. Existing server-side wrapper dependencies that iterate get_session(name) remain source-compatible. Migrate them when touched so FastAPI can propagate endpoint exceptions directly through rollback and cleanup. All partial(get_session, ...) forms are unsupported; replace them with session_dependency(name). In particular, a keyword-bound partial is rejected while the route is registered instead of silently selecting the wrong database. Never pass session_dependency itself to Depends; FastAPI would treat its name as a required request parameter and, if supplied, inject a callable instead of an AsyncSession.

A database name is trusted application wiring. Never derive it from a query parameter, path parameter, header, request body, or other request data.

FastAPI dependencies are needed only when the endpoint deliberately owns the unit of work or performs database work directly. If it owns the unit of work, pass the concrete injected session to participating lower layers. If a service or repository owns the unit of work, it should use db.session(name) or BaseSQLRepository.session() instead.

Default Database in FastAPI

from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from zodiac_core.db import get_session


@app.get("/users")
async def list_users(
    session: Annotated[AsyncSession, Depends(get_session)],
):
    ...

Named Database in FastAPI

Create the dependency while defining the router. This only binds a trusted name; it does not create a session or acquire a connection. The engine may be registered later during application lifespan. Store and reuse the returned callable: FastAPI's dependency cache and app.dependency_overrides identify it by callable identity. By default, repeated uses of that stored callable within one request share one session; a later request receives a new session. Passing DEFAULT_DB_NAME returns get_session; ordinary default-database routes should use get_session directly.

from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from zodiac_core.db import session_dependency


get_analytics_session = session_dependency("analytics")


@app.get("/reports")
async def list_reports(
    session: Annotated[AsyncSession, Depends(get_analytics_session)],
):
    ...

Compatibility with get_session(name)

The named-call API published in ZodiacCore 0.7.0 remains source- and call-compatible, including existing zero-argument wrapper dependencies:

async def legacy_analytics_session():
    async for session in get_session("analytics"):
        yield session

The name in this example is fixed by server code and cannot be replaced by a request parameter. Preserve the wrapper when compatibility requires it, but migrate touched routes and write new routes with session_dependency("analytics"); the returned dependency lets FastAPI propagate endpoint exceptions directly through session rollback and cleanup. Do not replace the wrapper with partial(get_session, ...); every partial form is unsupported. A keyword-bound partial is rejected during route registration because FastAPI dependency resolution would otherwise override its database name. Use session_dependency("analytics") instead.

For code outside FastAPI that owns a unit of work, use the context manager directly:

from zodiac_core.db import db


async def rebuild_reports() -> None:
    async with db.session("analytics") as session:
        ...
        await session.commit()

Do not add a route session dependency merely because the route eventually calls a repository. A BaseSQLRepository already owns its sessions by default; use a route dependency only when the route intentionally owns a shared unit of work or performs database work directly.

db.setup() creates one long-lived engine, connection pool, and async_sessionmaker for each named database. An AsyncSession is not kept in the connection pool: every dependency execution or db.session(...) context creates a new session. It normally borrows a connection lazily and returns it when the session closes. Never share an AsyncSession across requests or concurrent tasks.

5. Working with Repositories

Inherit from BaseSQLRepository to create your data access layer.

from sqlalchemy import select
from zodiac_core.db.repository import BaseSQLRepository

from .models import User


class UserRepository(BaseSQLRepository):
    async def find_by_username(self, username: str) -> User | None:
        async with self.session() as session:
            stmt = select(User).where(User.username == username)
            result = await session.execute(stmt)
            return result.scalar_one_or_none()

    async def create_user(self, user: User) -> User:
        async with self.session() as session:
            session.add(user)
            await session.commit()
            await session.refresh(user)
            return user

6. Multi-Database Support

ZodiacCore supports multiple database connections simultaneously. This is essential for architectures involving:

  • Read-Write Splitting: Routing writes to a Master and reads to a Replica.
  • Vertical Partitioning: Storing different modules (e.g., Users, Analytics) in separate databases.

Registering Named Databases

You can call db.setup() multiple times with different name arguments.

# Primary Database (Master)
db.setup("postgresql+asyncpg://master_db_url", name="default")

# Read-only Replica
db.setup("postgresql+asyncpg://replica_db_url", name="read_only")

Releasing Named Databases

Named shutdown is the companion to named setup:

from zodiac_core.db import db


async def shutdown_named_databases() -> None:
    # Dispose only the replica pool
    await db.shutdown(name="read_only")

    # Dispose everything registered in the manager
    await db.shutdown()

Use named shutdown when the process keeps other databases alive, such as multi-app hosting, plugin-based services, or multiple DI resources sharing the same global manager.

Binding Repositories to a Database

When creating a repository, specify which database it should use via db_name.

from zodiac_core.db.repository import BaseSQLRepository


class ReadOnlyUserRepository(BaseSQLRepository):
    def __init__(self) -> None:
        # This repo will always use the 'read_only' engine
        super().__init__(db_name="read_only")

    async def get_total_users(self) -> int:
        async with self.session() as session:
            # Executes on replica
            ...

7. API Reference

Session & Lifecycle

zodiac_core.db.session

DEFAULT_DB_NAME = 'default' module-attribute
db = DatabaseManager() module-attribute
DatabaseManager

Manages multiple Async Database Engines and Session Factories. Implemented as a Strict Singleton to coordinate connection pools.

Integration Examples:

  1. Native FastAPI (Lifespan + Depends):

    # main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from zodiac_core.db.session import db, get_session

@asynccontextmanager
async def lifespan(app: FastAPI):
    db.setup("sqlite+aiosqlite:///database.db")
    yield
    await db.shutdown()

app = FastAPI(lifespan=lifespan)

@app.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)):
    return {"status": "ok"}

  2. Dependency Injector (Using provided init_db_resource):

    # containers.py
from dependency_injector import containers, providers
from zodiac_core.utils import strtobool
from zodiac_core.db.session import init_db_resource

class Container(containers.DeclarativeContainer):
    config = providers.Configuration(strict=True)

    # Use the pre-built resource helper
    db_manager = providers.Resource(
        init_db_resource,
        database_url=config.db.url,
        echo=config.db.echo.as_(strtobool),
    )
Source code in zodiac_core/db/session.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
class DatabaseManager:
    """
    Manages multiple Async Database Engines and Session Factories.
    Implemented as a Strict Singleton to coordinate connection pools.

    Integration Examples:

    1. **Native FastAPI (Lifespan + Depends):**

        ```python
        # main.py
        from contextlib import asynccontextmanager
        from fastapi import FastAPI, Depends
        from sqlalchemy.ext.asyncio import AsyncSession
        from zodiac_core.db.session import db, get_session

        @asynccontextmanager
        async def lifespan(app: FastAPI):
            db.setup("sqlite+aiosqlite:///database.db")
            yield
            await db.shutdown()

        app = FastAPI(lifespan=lifespan)

        @app.get("/items")
        async def list_items(session: AsyncSession = Depends(get_session)):
            return {"status": "ok"}
        ```

    2. **Dependency Injector (Using provided init_db_resource):**

        ```python
        # containers.py
        from dependency_injector import containers, providers
        from zodiac_core.utils import strtobool
        from zodiac_core.db.session import init_db_resource

        class Container(containers.DeclarativeContainer):
            config = providers.Configuration(strict=True)

            # Use the pre-built resource helper
            db_manager = providers.Resource(
                init_db_resource,
                database_url=config.db.url,
                echo=config.db.echo.as_(strtobool),
            )
        ```
    """

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._engines: Dict[str, AsyncEngine] = {}
            cls._instance._session_factories: Dict[str, async_sessionmaker[AsyncSession]] = {}
            cls._instance._setup_configs: Dict[str, Dict[str, Any]] = {}
        return cls._instance

    def get_engine(self, name: str = DEFAULT_DB_NAME) -> AsyncEngine:
        """Access a specific SQLAlchemy AsyncEngine by name."""
        if name not in self._engines:
            raise RuntimeError(f"Database engine '{name}' is not initialized. Call db.setup(name='{name}') first.")
        return self._engines[name]

    def get_factory(self, name: str = DEFAULT_DB_NAME) -> async_sessionmaker[AsyncSession]:
        """Access a specific AsyncSession factory by name."""
        if name not in self._session_factories:
            raise RuntimeError(f"Session factory for '{name}' is not initialized. Call db.setup(name='{name}') first.")
        return self._session_factories[name]

    @property
    def engine(self) -> AsyncEngine:
        """Access the default SQLAlchemy AsyncEngine."""
        return self.get_engine(DEFAULT_DB_NAME)

    @property
    def session_factory(self) -> async_sessionmaker[AsyncSession]:
        """Access the default AsyncSession factory."""
        return self.get_factory(DEFAULT_DB_NAME)

    def setup(
        self,
        database_url: str,
        name: str = DEFAULT_DB_NAME,
        echo: bool = False,
        pool_size: int = 10,
        max_overflow: int = 20,
        pool_pre_ping: bool = True,
        connect_args: Optional[dict] = None,
        **kwargs,
    ) -> None:
        """Initialize an Async Engine and Session Factory with a specific name."""
        engine_args = {
            "echo": echo,
            "pool_pre_ping": pool_pre_ping,
            "connect_args": connect_args or {},
            **kwargs,
        }

        if "sqlite" not in database_url:
            engine_args["pool_size"] = pool_size
            engine_args["max_overflow"] = max_overflow

        current = {
            "database_url": database_url,
            "engine_args": deepcopy(engine_args),
        }

        if name in self._engines:
            existing = self._setup_configs.get(name)
            if existing == current:
                logger.debug(f"Database '{name}' is already configured with the same settings, skipping.")
                return
            raise RuntimeError(f"Database '{name}' is already configured with different settings")

        engine = create_async_engine(database_url, **engine_args)
        factory = async_sessionmaker(
            bind=engine,
            class_=AsyncSession,
            expire_on_commit=False,
            autoflush=False,
        )

        self._engines[name] = engine
        self._session_factories[name] = factory
        self._setup_configs[name] = current
        logger.info(f"Database '{name}' initialized successfully.")

    async def shutdown(self, name: str | None = None) -> None:
        """
        Dispose database resources.

        Args:
            name: Optional database name. When provided, only that engine/factory
                  is disposed. When omitted, all registered databases are disposed.
        """
        if name is not None:
            engine = self._engines.pop(name, None)
            self._session_factories.pop(name, None)
            self._setup_configs.pop(name, None)
            if engine is not None:
                await engine.dispose()
            return

        for engine in self._engines.values():
            await engine.dispose()
        self._engines.clear()
        self._session_factories.clear()
        self._setup_configs.clear()

    @asynccontextmanager
    async def session(self, name: str = DEFAULT_DB_NAME) -> AsyncGenerator[AsyncSession, None]:
        """Open a managed session for code that owns a unit of work.

        This is the general lifecycle API for background jobs, CLI commands,
        startup tasks, and services or repositories that own their unit of
        work. At the FastAPI boundary, ``get_session`` or a dependency returned
        by ``session_dependency`` may instead make the endpoint own the unit of
        work. Lower layers must never call those FastAPI dependency callables.
        If the endpoint owns the unit of work, pass the concrete
        ``AsyncSession`` to participating services and repositories; otherwise
        let the service or repository use this context manager or
        ``BaseSQLRepository.session()``.

        Each context creates a new ``AsyncSession`` and always closes it. The
        session borrows a connection lazily from the selected engine's pool;
        closing the session returns that connection to the pool. Exceptions
        trigger a rollback. Successful exit does not commit automatically.

        Note:
            This context manager does NOT auto-commit. You must explicitly call
            `await session.commit()` to persist changes to the database.

        Example:
            ```python
            async with db.session() as session:
                session.add(user)
                await session.commit()  # Required to persist changes
            ```
        """
        async with manage_session(self.get_factory(name)) as session:
            yield session

    async def verify(self, name: str = DEFAULT_DB_NAME) -> bool:
        """
        Verify the database connection is working.

        Args:
            name: The database name to verify.

        Returns:
            True if connection is successful.

        Raises:
            RuntimeError: If the database is not initialized.
            Exception: If the connection test fails.
        """
        async with self.session(name) as session:
            await session.execute(text("SELECT 1"))
        logger.info(f"Database '{name}' connection verified.")
        return True

    async def create_all(self, name: str = DEFAULT_DB_NAME, metadata: Any = None) -> None:
        """
        Create tables in the database.

        Args:
            name: The database name to create tables in.
            metadata: SQLAlchemy MetaData object. If None, uses SQLModel.metadata
                      which includes ALL registered models. For production, consider
                      using Alembic migrations instead.

        Example:
            ```python
            # Development: create all tables
            await db.create_all()

            # With custom metadata (only specific tables)
            from sqlalchemy import MetaData
            my_metadata = MetaData()
            await db.create_all(metadata=my_metadata)
            ```
        """
        target_metadata = metadata if metadata is not None else SQLModel.metadata
        async with self.get_engine(name).begin() as conn:
            await conn.run_sync(target_metadata.create_all)
engine property

Access the default SQLAlchemy AsyncEngine.

session_factory property

Access the default AsyncSession factory.

create_all(name=DEFAULT_DB_NAME, metadata=None) async

Create tables in the database.

Parameters:

Name Type Description Default
name str

The database name to create tables in.

 DEFAULT_DB_NAME
metadata Any

SQLAlchemy MetaData object. If None, uses SQLModel.metadata which includes ALL registered models. For production, consider using Alembic migrations instead.

 None
Example 
# Development: create all tables
await db.create_all()

# With custom metadata (only specific tables)
from sqlalchemy import MetaData
my_metadata = MetaData()
await db.create_all(metadata=my_metadata)
Source code in zodiac_core/db/session.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
async def create_all(self, name: str = DEFAULT_DB_NAME, metadata: Any = None) -> None:
    """
    Create tables in the database.

    Args:
        name: The database name to create tables in.
        metadata: SQLAlchemy MetaData object. If None, uses SQLModel.metadata
                  which includes ALL registered models. For production, consider
                  using Alembic migrations instead.

    Example:
        ```python
        # Development: create all tables
        await db.create_all()

        # With custom metadata (only specific tables)
        from sqlalchemy import MetaData
        my_metadata = MetaData()
        await db.create_all(metadata=my_metadata)
        ```
    """
    target_metadata = metadata if metadata is not None else SQLModel.metadata
    async with self.get_engine(name).begin() as conn:
        await conn.run_sync(target_metadata.create_all)
get_engine(name=DEFAULT_DB_NAME)

Access a specific SQLAlchemy AsyncEngine by name.

Source code in zodiac_core/db/session.py 
114
115
116
117
118
def get_engine(self, name: str = DEFAULT_DB_NAME) -> AsyncEngine:
    """Access a specific SQLAlchemy AsyncEngine by name."""
    if name not in self._engines:
        raise RuntimeError(f"Database engine '{name}' is not initialized. Call db.setup(name='{name}') first.")
    return self._engines[name]
get_factory(name=DEFAULT_DB_NAME)

Access a specific AsyncSession factory by name.

Source code in zodiac_core/db/session.py 
120
121
122
123
124
def get_factory(self, name: str = DEFAULT_DB_NAME) -> async_sessionmaker[AsyncSession]:
    """Access a specific AsyncSession factory by name."""
    if name not in self._session_factories:
        raise RuntimeError(f"Session factory for '{name}' is not initialized. Call db.setup(name='{name}') first.")
    return self._session_factories[name]
session(name=DEFAULT_DB_NAME) async

Open a managed session for code that owns a unit of work.

This is the general lifecycle API for background jobs, CLI commands, startup tasks, and services or repositories that own their unit of work. At the FastAPI boundary, get_session or a dependency returned by session_dependency may instead make the endpoint own the unit of work. Lower layers must never call those FastAPI dependency callables. If the endpoint owns the unit of work, pass the concrete AsyncSession to participating services and repositories; otherwise let the service or repository use this context manager or BaseSQLRepository.session().

Each context creates a new AsyncSession and always closes it. The session borrows a connection lazily from the selected engine's pool; closing the session returns that connection to the pool. Exceptions trigger a rollback. Successful exit does not commit automatically.

Note

This context manager does NOT auto-commit. You must explicitly call await session.commit() to persist changes to the database.

Example 
async with db.session() as session:
    session.add(user)
    await session.commit()  # Required to persist changes
Source code in zodiac_core/db/session.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@asynccontextmanager
async def session(self, name: str = DEFAULT_DB_NAME) -> AsyncGenerator[AsyncSession, None]:
    """Open a managed session for code that owns a unit of work.

    This is the general lifecycle API for background jobs, CLI commands,
    startup tasks, and services or repositories that own their unit of
    work. At the FastAPI boundary, ``get_session`` or a dependency returned
    by ``session_dependency`` may instead make the endpoint own the unit of
    work. Lower layers must never call those FastAPI dependency callables.
    If the endpoint owns the unit of work, pass the concrete
    ``AsyncSession`` to participating services and repositories; otherwise
    let the service or repository use this context manager or
    ``BaseSQLRepository.session()``.

    Each context creates a new ``AsyncSession`` and always closes it. The
    session borrows a connection lazily from the selected engine's pool;
    closing the session returns that connection to the pool. Exceptions
    trigger a rollback. Successful exit does not commit automatically.

    Note:
        This context manager does NOT auto-commit. You must explicitly call
        `await session.commit()` to persist changes to the database.

    Example:
        ```python
        async with db.session() as session:
            session.add(user)
            await session.commit()  # Required to persist changes
        ```
    """
    async with manage_session(self.get_factory(name)) as session:
        yield session
setup(database_url, name=DEFAULT_DB_NAME, echo=False, pool_size=10, max_overflow=20, pool_pre_ping=True, connect_args=None, **kwargs)

Initialize an Async Engine and Session Factory with a specific name.

Source code in zodiac_core/db/session.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def setup(
    self,
    database_url: str,
    name: str = DEFAULT_DB_NAME,
    echo: bool = False,
    pool_size: int = 10,
    max_overflow: int = 20,
    pool_pre_ping: bool = True,
    connect_args: Optional[dict] = None,
    **kwargs,
) -> None:
    """Initialize an Async Engine and Session Factory with a specific name."""
    engine_args = {
        "echo": echo,
        "pool_pre_ping": pool_pre_ping,
        "connect_args": connect_args or {},
        **kwargs,
    }

    if "sqlite" not in database_url:
        engine_args["pool_size"] = pool_size
        engine_args["max_overflow"] = max_overflow

    current = {
        "database_url": database_url,
        "engine_args": deepcopy(engine_args),
    }

    if name in self._engines:
        existing = self._setup_configs.get(name)
        if existing == current:
            logger.debug(f"Database '{name}' is already configured with the same settings, skipping.")
            return
        raise RuntimeError(f"Database '{name}' is already configured with different settings")

    engine = create_async_engine(database_url, **engine_args)
    factory = async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=False,
        autoflush=False,
    )

    self._engines[name] = engine
    self._session_factories[name] = factory
    self._setup_configs[name] = current
    logger.info(f"Database '{name}' initialized successfully.")
shutdown(name=None) async

Dispose database resources.

Parameters:

Name Type Description Default
name str | None

Optional database name. When provided, only that engine/factory is disposed. When omitted, all registered databases are disposed.

 None
Source code in zodiac_core/db/session.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
async def shutdown(self, name: str | None = None) -> None:
    """
    Dispose database resources.

    Args:
        name: Optional database name. When provided, only that engine/factory
              is disposed. When omitted, all registered databases are disposed.
    """
    if name is not None:
        engine = self._engines.pop(name, None)
        self._session_factories.pop(name, None)
        self._setup_configs.pop(name, None)
        if engine is not None:
            await engine.dispose()
        return

    for engine in self._engines.values():
        await engine.dispose()
    self._engines.clear()
    self._session_factories.clear()
    self._setup_configs.clear()
verify(name=DEFAULT_DB_NAME) async

Verify the database connection is working.

Parameters:

Name Type Description Default
name str

The database name to verify.

 DEFAULT_DB_NAME

Returns:

Type Description
bool

True if connection is successful.

Raises:

Type Description
RuntimeError

If the database is not initialized.
Exception

If the connection test fails.
Source code in zodiac_core/db/session.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def verify(self, name: str = DEFAULT_DB_NAME) -> bool:
    """
    Verify the database connection is working.

    Args:
        name: The database name to verify.

    Returns:
        True if connection is successful.

    Raises:
        RuntimeError: If the database is not initialized.
        Exception: If the connection test fails.
    """
    async with self.session(name) as session:
        await session.execute(text("SELECT 1"))
    logger.info(f"Database '{name}' connection verified.")
    return True
get_session(name=_DEFAULT_DATABASE_NAME_PARAMETER) async

Yield a managed session while preserving the public named-call API.

At the FastAPI API/DI boundary, use Depends(get_session) only for the default database. FastAPI resolves name through a private dependency, so it is not exposed as request input.

The optional name argument remains source- and call-compatible with the public API introduced in ZodiacCore 0.7.0, including existing server-side wrapper dependencies that iterate get_session("analytics"). New named FastAPI wiring should use session_dependency(name) so FastAPI directly manages exception propagation and cleanup. Never use partial(get_session, ...). Keyword-bound partials are rejected during route registration because FastAPI would otherwise override the bound database name while resolving dependencies.

Services, repositories, jobs, and CLI commands should not treat this FastAPI-oriented generator as their general session API. If the endpoint owns a unit of work, pass its concrete AsyncSession to participating lower layers. Otherwise use db.session(...) or BaseSQLRepository.session() at the layer that owns the unit of work.

By default, FastAPI resolves this dependency callable once per request, so repeated uses in that request share one AsyncSession; a later request gets a new session. The session is not a pooled connection: it borrows a connection lazily from the selected engine's pool and returns it during cleanup.

Note

This dependency does NOT auto-commit. You must explicitly call await session.commit() within your endpoint to persist changes.

Example 
# Default database — use directly as a dependency
@app.post("/users")
async def create_user(session: AsyncSession = Depends(get_session)):
    session.add(User(name="test"))
    await session.commit()
    return user
Source code in zodiac_core/db/session.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
async def get_session(
    name: _ServerControlledDatabaseName = _DEFAULT_DATABASE_NAME_PARAMETER,
) -> AsyncGenerator[AsyncSession, None]:
    """Yield a managed session while preserving the public named-call API.

    At the FastAPI API/DI boundary, use ``Depends(get_session)`` only for the
    default database. FastAPI resolves ``name`` through a private dependency,
    so it is not exposed as request input.

    The optional ``name`` argument remains source- and call-compatible with the
    public API introduced in ZodiacCore 0.7.0, including existing server-side
    wrapper dependencies that iterate ``get_session("analytics")``. New named
    FastAPI wiring should use ``session_dependency(name)`` so FastAPI directly
    manages exception propagation and cleanup. Never use
    ``partial(get_session, ...)``. Keyword-bound partials are rejected during
    route registration because FastAPI would otherwise override the bound
    database name while resolving dependencies.

    Services, repositories, jobs, and CLI commands should not treat this
    FastAPI-oriented generator as their general session API. If the endpoint
    owns a unit of work, pass its concrete ``AsyncSession`` to participating
    lower layers. Otherwise use ``db.session(...)`` or
    ``BaseSQLRepository.session()`` at the layer that owns the unit of work.

    By default, FastAPI resolves this dependency callable once per request, so
    repeated uses in that request share one ``AsyncSession``; a later request
    gets a new session. The session is not a pooled connection: it borrows a
    connection lazily from the selected engine's pool and returns it during
    cleanup.

    Note:
        This dependency does NOT auto-commit. You must explicitly call
        `await session.commit()` within your endpoint to persist changes.

    Example:
        ```python
        # Default database — use directly as a dependency
        @app.post("/users")
        async def create_user(session: AsyncSession = Depends(get_session)):
            session.add(User(name="test"))
            await session.commit()
            return user
        ```
    """
    async with db.session(name) as session:
        yield session
session_dependency(name)

Create a request-parameter-free dependency bound to a named database.

This is a FastAPI dependency factory, not a session factory or context manager. Call it once while defining API wiring, then pass the returned callable to Depends. Calling this factory creates neither an AsyncSession nor a database connection. FastAPI creates the AsyncSession when it resolves the returned dependency; a pooled connection is normally checked out only when database work begins.

The database name must be fixed by server-side application code. Never derive it from a query parameter, path parameter, header, request body, or any other request data.

Do not pass session_dependency itself to Depends: FastAPI would treat name as a required request parameter and, if supplied, inject the returned callable instead of an AsyncSession. Do not use partial(get_session, ...); keyword-bound partials fail during route registration instead of risking a silent connection to the wrong database. Existing server-side wrapper dependencies that iterate get_session(name) remain source-compatible, but new named FastAPI routes should use this factory for direct exception propagation and cleanup. Do not manually call or iterate the returned dependency from business code; lower layers that own a unit of work use db.session(name) or BaseSQLRepository.session() instead.

Create and store one named dependency at module or router scope. FastAPI dependency caching and app.dependency_overrides identify dependencies by callable identity, so pass the stored callable everywhere instead of repeatedly calling this factory.

Parameters:

Name Type Description Default
name str

Database name registered with db.setup(..., name=name).

 required

Returns:

Type Description
Callable[[], AsyncGenerator[AsyncSession, None]]

A FastAPI-safe async-generator dependency. By default, FastAPI creates
Callable[[], AsyncGenerator[AsyncSession, None]]

one AsyncSession for this callable per request; repeated uses of the
Callable[[], AsyncGenerator[AsyncSession, None]]

stored callable share that session. Named dependencies are zero-argument
Callable[[], AsyncGenerator[AsyncSession, None]]

closures. Passing DEFAULT_DB_NAME returns get_session itself so
Callable[[], AsyncGenerator[AsyncSession, None]]

default-database dependency caching and overrides remain unified.
Example 
get_analytics_session = session_dependency("analytics")

@app.get("/reports")
async def get_reports(
    session: AsyncSession = Depends(get_analytics_session),
):
    ...
Source code in zodiac_core/db/session.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
def session_dependency(name: str) -> Callable[[], AsyncGenerator[AsyncSession, None]]:
    """Create a request-parameter-free dependency bound to a named database.

    This is a FastAPI dependency factory, not a session factory or context
    manager. Call it once while defining API wiring, then pass the returned
    callable to ``Depends``. Calling this factory creates neither an
    ``AsyncSession`` nor a database connection. FastAPI creates the
    ``AsyncSession`` when it resolves the returned dependency; a pooled
    connection is normally checked out only when database work begins.

    The database name must be fixed by server-side application code. Never
    derive it from a query parameter, path parameter, header, request body, or
    any other request data.

    Do not pass ``session_dependency`` itself to ``Depends``: FastAPI would
    treat ``name`` as a required request parameter and, if supplied, inject the
    returned callable instead of an ``AsyncSession``. Do not use
    ``partial(get_session, ...)``; keyword-bound partials fail during route
    registration instead of risking a silent connection to the wrong database.
    Existing server-side wrapper dependencies that iterate
    ``get_session(name)`` remain source-compatible, but new named FastAPI routes
    should use this factory for direct exception propagation and cleanup. Do not
    manually call or iterate the returned dependency from business code; lower
    layers that own a unit of work use ``db.session(name)`` or
    ``BaseSQLRepository.session()`` instead.

    Create and store one named dependency at module or router scope. FastAPI
    dependency caching and ``app.dependency_overrides`` identify dependencies
    by callable identity, so pass the stored callable everywhere instead of
    repeatedly calling this factory.

    Args:
        name: Database name registered with ``db.setup(..., name=name)``.

    Returns:
        A FastAPI-safe async-generator dependency. By default, FastAPI creates
        one ``AsyncSession`` for this callable per request; repeated uses of the
        stored callable share that session. Named dependencies are zero-argument
        closures. Passing ``DEFAULT_DB_NAME`` returns ``get_session`` itself so
        default-database dependency caching and overrides remain unified.

    Example:
        ```python
        get_analytics_session = session_dependency("analytics")

        @app.get("/reports")
        async def get_reports(
            session: AsyncSession = Depends(get_analytics_session),
        ):
            ...
        ```
    """
    if name == DEFAULT_DB_NAME:
        return get_session

    async def get_named_session() -> AsyncGenerator[AsyncSession, None]:
        async with db.session(name) as session:
            yield session

    return get_named_session
init_db_resource(database_url, name=DEFAULT_DB_NAME, echo=False, connect_args=None, **kwargs) async

A helper for dependency_injector's Resource provider. Handles the setup and shutdown lifecycle of the global db instance. Cleanup is scoped to the provided database name, so other registered databases remain available.

Source code in zodiac_core/db/session.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
async def init_db_resource(
    database_url: str,
    name: str = DEFAULT_DB_NAME,
    echo: bool = False,
    connect_args: Optional[dict] = None,
    **kwargs,
) -> AsyncGenerator[DatabaseManager, None]:
    """
    A helper for dependency_injector's Resource provider.
    Handles the setup and shutdown lifecycle of the global `db` instance.
    Cleanup is scoped to the provided database `name`, so other registered
    databases remain available.
    """
    db.setup(database_url=database_url, name=name, echo=echo, connect_args=connect_args, **kwargs)
    try:
        yield db
    finally:
        await db.shutdown(name=name)

Repository Base

zodiac_core.db.repository.BaseSQLRepository

Standard base class for SQL-based repositories.

Supports multiple database instances via db_name and provides professional utilities for common operations like pagination.

Source code in zodiac_core/db/repository.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
class BaseSQLRepository:
    """
    Standard base class for SQL-based repositories.

    Supports multiple database instances via `db_name` and provides
    professional utilities for common operations like pagination.
    """

    sort_spec: SortSpec | None = None

    def __init__(
        self,
        session_factory: Optional[async_sessionmaker[AsyncSession]] = None,
        db_name: str = DEFAULT_DB_NAME,
        options: Optional[Any] = None,
    ) -> None:
        """
        Initialize the repository.

        Args:
            session_factory: Optional custom session factory. If provided, db_name is ignored.
            db_name: The name of the database engine registered in db.setup(). Defaults to DEFAULT_DB_NAME ("default").
            options: Optional configuration/options for the repository.
        """
        self._session_factory = session_factory
        self.db_name = db_name
        self.options = options

    @asynccontextmanager
    async def session(self) -> AsyncIterator[AsyncSession]:
        """
        Async context manager for obtaining a database session.
        Uses the injected factory or resolves one from the global 'db' via 'db_name'.

        Note:
            This context manager does NOT auto-commit. You must explicitly call
            `await session.commit()` to persist changes to the database.
        """
        factory = self._session_factory or db.get_factory(self.db_name)
        async with manage_session(factory) as session:
            yield session

    async def paginate(
        self,
        session: AsyncSession,
        statement: Any,
        params: PageParams,
        transformer: Optional[Type[T]] = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> PagedResponse[T]:
        """
        Execute a paginated query with automatic count and paging.

        Performs:
        1. Optional multi-column sorting using public field mappings.
        2. Automatic total count query using the provided statement.
        3. Automatic limit/offset application.
        4. Packaging results into a standardized PagedResponse.

        Args:
            session: The active AsyncSession.
            statement: The SQLAlchemy select statement (without limit/offset).
            params: Standard PageParams (page, size).
            transformer: Optional Pydantic model to transform DB objects into.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            async with self.session() as session:
                stmt = select(UserModel).order_by(UserModel.created_at.desc())
                return await self.paginate(session, stmt, params)
            ```
        """
        effective_sort_spec = sort_spec or self.sort_spec

        if effective_sort_spec is not None:
            sort_params = params if isinstance(params, SortParams) else None
            statement = self.apply_sorting(statement, sort_params, sort_spec=effective_sort_spec)

        # 1. Execute Count Query
        # Remove limit/offset/order_by (if any) for count query, then wrap in subquery.
        # Wrapping in subquery handles complex queries (joins, groups).
        count_base = statement.limit(None).offset(None).order_by(None)
        count_stmt = select(func.count()).select_from(count_base.subquery())
        total = (await session.execute(count_stmt)).scalar() or 0

        # 2. Execute Paged Query
        skip = (params.page - 1) * params.size
        paged_stmt = statement.offset(skip).limit(params.size)
        result = await session.execute(paged_stmt)
        items = result.scalars().all()

        # 3. Optional Transformation
        if transformer:
            items = [transformer.model_validate(item) for item in items]

        return PagedResponse.create(items=list(items), total=total, params=params)

    def apply_sorting(
        self,
        statement: Any,
        sort_params: SortParams | None = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> Any:
        """
        Apply validated multi-column sorting to a SQLAlchemy statement.
        Existing ORDER BY clauses are replaced when sort fields are present.

        Args:
            statement: The SQLAlchemy select statement to sort.
            sort_params: Standard SortParams or PageSortParams. When omitted,
                only the configured default sort is applied.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            stmt = self.apply_sorting(
                select(UserModel),
                params,
                sort_spec=SortSpec(
                    columns={
                        "name": UserModel.name,
                        "created_at": UserModel.created_at,
                    }
                ),
            )
            ```
        """
        if sort_params is not None and not isinstance(sort_params, SortParams):
            raise TypeError("sort_params must be SortParams or PageSortParams")

        effective_sort_spec = sort_spec or self.sort_spec

        if effective_sort_spec is None:
            return statement

        sort_pairs = effective_sort_spec.pairs_for(sort_params)
        if not sort_pairs:
            return statement

        order_by_clauses = []
        for field, direction in sort_pairs:
            column = effective_sort_spec.columns.get(field)
            if column is None:
                supported_fields = sorted(effective_sort_spec.columns)
                raise BadRequestException(
                    message=(
                        f"Unsupported sort field '{field}'. Supported sort fields: {', '.join(supported_fields)}."
                    ),
                    data={
                        "field": field,
                        "supported_fields": supported_fields,
                    },
                )
            order_by_clauses.append(column.asc() if direction == "asc" else column.desc())

        return statement.order_by(None).order_by(*order_by_clauses)

    async def paginate_query(
        self,
        statement: Any,
        params: PageParams,
        transformer: Optional[Type[T]] = None,
        *,
        sort_spec: SortSpec | None = None,
    ) -> PagedResponse[T]:
        """
        Convenience method that automatically manages session for pagination.

        This is a wrapper around `paginate()` that handles session management,
        making it easier to use in repository methods.

        Args:
            statement: The SQLAlchemy select statement (without limit/offset).
            params: Standard PageParams (page, size).
            transformer: Optional Pydantic model to transform DB objects into.
            sort_spec: Optional reusable sort configuration. When omitted, the
                repository class-level ``sort_spec`` is used.

        Example:
            ```python
            async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
                stmt = select(ItemModel).order_by(ItemModel.id)
                return await self.paginate_query(stmt, params)
            ```
        """
        async with self.session() as session:
            return await self.paginate(
                session,
                statement,
                params,
                transformer,
                sort_spec=sort_spec,
            )
__init__(session_factory=None, db_name=DEFAULT_DB_NAME, options=None)

Initialize the repository.

Parameters:

Name Type Description Default
session_factory Optional[async_sessionmaker[AsyncSession]]

Optional custom session factory. If provided, db_name is ignored.

 None
db_name str

The name of the database engine registered in db.setup(). Defaults to DEFAULT_DB_NAME ("default").

 DEFAULT_DB_NAME
options Optional[Any]

Optional configuration/options for the repository.

 None
Source code in zodiac_core/db/repository.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    session_factory: Optional[async_sessionmaker[AsyncSession]] = None,
    db_name: str = DEFAULT_DB_NAME,
    options: Optional[Any] = None,
) -> None:
    """
    Initialize the repository.

    Args:
        session_factory: Optional custom session factory. If provided, db_name is ignored.
        db_name: The name of the database engine registered in db.setup(). Defaults to DEFAULT_DB_NAME ("default").
        options: Optional configuration/options for the repository.
    """
    self._session_factory = session_factory
    self.db_name = db_name
    self.options = options
apply_sorting(statement, sort_params=None, *, sort_spec=None)

Apply validated multi-column sorting to a SQLAlchemy statement. Existing ORDER BY clauses are replaced when sort fields are present.

Parameters:

Name Type Description Default
statement Any

The SQLAlchemy select statement to sort.

 required
sort_params SortParams | None

Standard SortParams or PageSortParams. When omitted, only the configured default sort is applied.

 None
sort_spec SortSpec | None

Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.

 None
Example 
stmt = self.apply_sorting(
    select(UserModel),
    params,
    sort_spec=SortSpec(
        columns={
            "name": UserModel.name,
            "created_at": UserModel.created_at,
        }
    ),
)
Source code in zodiac_core/db/repository.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def apply_sorting(
    self,
    statement: Any,
    sort_params: SortParams | None = None,
    *,
    sort_spec: SortSpec | None = None,
) -> Any:
    """
    Apply validated multi-column sorting to a SQLAlchemy statement.
    Existing ORDER BY clauses are replaced when sort fields are present.

    Args:
        statement: The SQLAlchemy select statement to sort.
        sort_params: Standard SortParams or PageSortParams. When omitted,
            only the configured default sort is applied.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        stmt = self.apply_sorting(
            select(UserModel),
            params,
            sort_spec=SortSpec(
                columns={
                    "name": UserModel.name,
                    "created_at": UserModel.created_at,
                }
            ),
        )
        ```
    """
    if sort_params is not None and not isinstance(sort_params, SortParams):
        raise TypeError("sort_params must be SortParams or PageSortParams")

    effective_sort_spec = sort_spec or self.sort_spec

    if effective_sort_spec is None:
        return statement

    sort_pairs = effective_sort_spec.pairs_for(sort_params)
    if not sort_pairs:
        return statement

    order_by_clauses = []
    for field, direction in sort_pairs:
        column = effective_sort_spec.columns.get(field)
        if column is None:
            supported_fields = sorted(effective_sort_spec.columns)
            raise BadRequestException(
                message=(
                    f"Unsupported sort field '{field}'. Supported sort fields: {', '.join(supported_fields)}."
                ),
                data={
                    "field": field,
                    "supported_fields": supported_fields,
                },
            )
        order_by_clauses.append(column.asc() if direction == "asc" else column.desc())

    return statement.order_by(None).order_by(*order_by_clauses)
paginate(session, statement, params, transformer=None, *, sort_spec=None) async

Execute a paginated query with automatic count and paging.

Performs: 1. Optional multi-column sorting using public field mappings. 2. Automatic total count query using the provided statement. 3. Automatic limit/offset application. 4. Packaging results into a standardized PagedResponse.

Parameters:

Name Type Description Default
session AsyncSession

The active AsyncSession.

 required
statement Any

The SQLAlchemy select statement (without limit/offset).

 required
params PageParams

Standard PageParams (page, size).

 required
transformer Optional[Type[T]]

Optional Pydantic model to transform DB objects into.

 None
sort_spec SortSpec | None

Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.

 None
Example 
async with self.session() as session:
    stmt = select(UserModel).order_by(UserModel.created_at.desc())
    return await self.paginate(session, stmt, params)
Source code in zodiac_core/db/repository.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def paginate(
    self,
    session: AsyncSession,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = None,
    *,
    sort_spec: SortSpec | None = None,
) -> PagedResponse[T]:
    """
    Execute a paginated query with automatic count and paging.

    Performs:
    1. Optional multi-column sorting using public field mappings.
    2. Automatic total count query using the provided statement.
    3. Automatic limit/offset application.
    4. Packaging results into a standardized PagedResponse.

    Args:
        session: The active AsyncSession.
        statement: The SQLAlchemy select statement (without limit/offset).
        params: Standard PageParams (page, size).
        transformer: Optional Pydantic model to transform DB objects into.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        async with self.session() as session:
            stmt = select(UserModel).order_by(UserModel.created_at.desc())
            return await self.paginate(session, stmt, params)
        ```
    """
    effective_sort_spec = sort_spec or self.sort_spec

    if effective_sort_spec is not None:
        sort_params = params if isinstance(params, SortParams) else None
        statement = self.apply_sorting(statement, sort_params, sort_spec=effective_sort_spec)

    # 1. Execute Count Query
    # Remove limit/offset/order_by (if any) for count query, then wrap in subquery.
    # Wrapping in subquery handles complex queries (joins, groups).
    count_base = statement.limit(None).offset(None).order_by(None)
    count_stmt = select(func.count()).select_from(count_base.subquery())
    total = (await session.execute(count_stmt)).scalar() or 0

    # 2. Execute Paged Query
    skip = (params.page - 1) * params.size
    paged_stmt = statement.offset(skip).limit(params.size)
    result = await session.execute(paged_stmt)
    items = result.scalars().all()

    # 3. Optional Transformation
    if transformer:
        items = [transformer.model_validate(item) for item in items]

    return PagedResponse.create(items=list(items), total=total, params=params)
paginate_query(statement, params, transformer=None, *, sort_spec=None) async

Convenience method that automatically manages session for pagination.

This is a wrapper around paginate() that handles session management, making it easier to use in repository methods.

Parameters:

Name Type Description Default
statement Any

The SQLAlchemy select statement (without limit/offset).

 required
params PageParams

Standard PageParams (page, size).

 required
transformer Optional[Type[T]]

Optional Pydantic model to transform DB objects into.

 None
sort_spec SortSpec | None

Optional reusable sort configuration. When omitted, the repository class-level sort_spec is used.

 None
Example 
async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
    stmt = select(ItemModel).order_by(ItemModel.id)
    return await self.paginate_query(stmt, params)
Source code in zodiac_core/db/repository.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
async def paginate_query(
    self,
    statement: Any,
    params: PageParams,
    transformer: Optional[Type[T]] = None,
    *,
    sort_spec: SortSpec | None = None,
) -> PagedResponse[T]:
    """
    Convenience method that automatically manages session for pagination.

    This is a wrapper around `paginate()` that handles session management,
    making it easier to use in repository methods.

    Args:
        statement: The SQLAlchemy select statement (without limit/offset).
        params: Standard PageParams (page, size).
        transformer: Optional Pydantic model to transform DB objects into.
        sort_spec: Optional reusable sort configuration. When omitted, the
            repository class-level ``sort_spec`` is used.

    Example:
        ```python
        async def list_items(self, params: PageParams) -> PagedResponse[ItemModel]:
            stmt = select(ItemModel).order_by(ItemModel.id)
            return await self.paginate_query(stmt, params)
        ```
    """
    async with self.session() as session:
        return await self.paginate(
            session,
            statement,
            params,
            transformer,
            sort_spec=sort_spec,
        )
session() async

Async context manager for obtaining a database session. Uses the injected factory or resolves one from the global 'db' via 'db_name'.

Note

This context manager does NOT auto-commit. You must explicitly call await session.commit() to persist changes to the database.

Source code in zodiac_core/db/repository.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
@asynccontextmanager
async def session(self) -> AsyncIterator[AsyncSession]:
    """
    Async context manager for obtaining a database session.
    Uses the injected factory or resolves one from the global 'db' via 'db_name'.

    Note:
        This context manager does NOT auto-commit. You must explicitly call
        `await session.commit()` to persist changes to the database.
    """
    factory = self._session_factory or db.get_factory(self.db_name)
    async with manage_session(factory) as session:
        yield session

SQL Models & Mixins

zodiac_core.db.sql

IntIDModel

Bases: SQLBase, IntIDMixin

Base SQLModel with Integer ID and Timestamps. Includes: ID (int) + CreatedAt + UpdatedAt.

Source code in zodiac_core/db/sql.py 
102
103
104
105
106
class IntIDModel(SQLBase, IntIDMixin):
    """
    Base SQLModel with Integer ID and Timestamps.
    Includes: ID (int) + CreatedAt + UpdatedAt.
    """
UUIDModel

Bases: SQLBase, UUIDMixin

Base SQLModel with UUID and Timestamps. Includes: ID (UUID) + CreatedAt + UpdatedAt.

Source code in zodiac_core/db/sql.py 
109
110
111
112
113
class UUIDModel(SQLBase, UUIDMixin):
    """
    Base SQLModel with UUID and Timestamps.
    Includes: ID (UUID) + CreatedAt + UpdatedAt.
    """
SQLDateTimeMixin

Bases: SQLModel

Mixin for created_at and updated_at with SQLAlchemy server-side defaults. Supports PostgreSQL, MySQL, and SQLite with proper UTC handling.

Source code in zodiac_core/db/sql.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class SQLDateTimeMixin(SQLModel):
    """
    Mixin for created_at and updated_at with SQLAlchemy server-side defaults.
    Supports PostgreSQL, MySQL, and SQLite with proper UTC handling.
    """

    created_at: datetime = Field(
        default_factory=utc_now,
        sa_column_kwargs={
            "server_default": utcnow(),
            "nullable": False,
        },
        sa_type=DateTime(timezone=True),
        description="Record creation timestamp (UTC)",
    )
    updated_at: datetime = Field(
        default_factory=utc_now,
        sa_column_kwargs={
            "server_default": utcnow(),
            "onupdate": utcnow(),
            "nullable": False,
        },
        sa_type=DateTime(timezone=True),
        description="Record last update timestamp (UTC)",
    )