Database & Migrations

Async SQLAlchemy engine configuration, connection pooling, model definitions, the CRUD mixin, Alembic migration workflow, and script execution tracking.

Singularity uses async SQLAlchemy with PostgreSQL as its primary database. The database layer is configured in singularity/db/ and provides both async sessions (for FastAPI request handling) and sync sessions (for Celery background tasks). Migrations are managed with Alembic, and startup scripts are tracked via a dedicated script_run_tracker table.

Database Connection Flow

The following diagram shows how requests flow from the application through the connection pool to PostgreSQL, and how Alembic operates on the database independently:

Two separate engines exist because FastAPI runs an async event loop (requiring asyncpg) while Celery workers run synchronous code (requiring psycopg2). Both connect to the same PostgreSQL database.

Async SQLAlchemy Setup

The database configuration lives in singularity/db/postgres.py. It creates both async and sync engines from the same database URL in settings:

# singularity/db/engine.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import AsyncAdaptedQueuePool, QueuePool
from singularity.config import settings

# Async engine (for FastAPI endpoints)
async_engine = create_async_engine(
    settings.database_url,                  # postgresql+asyncpg://...
    echo=False,
    poolclass=AsyncAdaptedQueuePool,
    pool_size=20,
    pool_timeout=3600,
    pool_pre_ping=True,
)

async_session = async_sessionmaker[AsyncSession](
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Sync engine (for Celery tasks)
sync_database_url = settings.database_url.replace(
    "postgresql+asyncpg://", "postgresql://"
)
sync_engine = create_engine(
    sync_database_url,
    echo=False,
    poolclass=QueuePool,
    pool_size=100,
    max_overflow=200,
    pool_timeout=30,
    pool_pre_ping=True,
)

sync_session = sessionmaker[Session](
    sync_engine,
    class_=Session,
    expire_on_commit=False,
)

Connection Pool Configuration

Parameter	Async Engine	Sync Engine	Description
`poolclass`	`AsyncAdaptedQueuePool`	`QueuePool`	Connection pool implementation
`pool_size`	`20`	`100`	Number of persistent connections
`max_overflow`	default	`200`	Extra connections beyond `pool_size`
`pool_timeout`	`3600`	`30`	Seconds to wait for a connection from the pool
`pool_pre_ping`	`True`	`True`	Test connections before use (prevents stale connections)
`expire_on_commit`	`False`	`False`	Keep object attributes accessible after commit

The async engine uses a large pool_timeout (3600s) because FastAPI handles many concurrent requests that may hold sessions open during long operations. The sync engine uses a shorter timeout because Celery tasks are typically shorter-lived and should fail fast if the pool is exhausted.

The `get_db` Dependency

For FastAPI endpoints that need a database session outside the service layer (e.g., direct router dependencies), the get_db async generator is available:

# singularity/db/engine.py
async def get_db():
    async with async_session() as session:
        yield session

Use it as a FastAPI dependency:

from fastapi import Depends
from singularity.db import get_db

@router.get("/direct")
async def direct_query(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

However, most services access the database through the Acquire container instead, which provides the same async_session factory.

Model Definitions

Models are defined in singularity/db/models.py using SQLAlchemy's declarative mapping. The module provides a Base class, utility mixins for common columns, and a CRUD abstract base that adds built-in create/read/update/delete methods.

Base Class

All models inherit from Base, which is a standard SQLAlchemy DeclarativeBase:

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Utility Mixins

Two mixins provide common column patterns:

import uuid as uuid_pkg
from sqlalchemy import DateTime, text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column

class UUIDMixin:
    """Adds a UUID primary key with server-side default."""
    id: Mapped[uuid_pkg.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )

class TimestampMixin:
    """Adds a created_at timestamp with server-side default."""
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
        server_default=text("CURRENT_TIMESTAMP"),
    )

The CRUD Mixin

The CRUD class combines both mixins and adds instance methods for common database operations. It is declared as __abstract__ = True so SQLAlchemy does not create a table for it directly:

class CRUD(Base, UUIDMixin, TimestampMixin):
    __abstract__ = True

    async def create(self):
        async with async_session() as session:
            session.add(self)
            await session.commit()
            return self

    @classmethod
    async def read(cls, _id: uuid_pkg.UUID):
        async with async_session() as session:
            query = select(cls).where(cls.id == _id)
            result = await session.execute(query)
            return result.scalars().first()

    @classmethod
    async def read_all(cls, **kwargs):
        async with async_session() as session:
            if kwargs:
                filters = [getattr(cls, key) == value
                           for key, value in kwargs.items()]
                query = select(cls).where(*filters)
            else:
                query = select(cls)
            result = await session.execute(query)
            return result.scalars().all()

    async def update(self, **kwargs):
        async with async_session() as session:
            for attr, value in kwargs.items():
                setattr(self, attr, value)
            session.add(self)
            await session.commit()
            return self

    async def delete(self):
        async with async_session() as session:
            await session.delete(self)
            await session.commit()

Defining a Model

To create a new model, inherit from CRUD (for full CRUD operations) or Base (for a plain model):

# services/v1/users/models.py
from sqlalchemy import Column, String, Boolean
from singularity.db.models import CRUD

class User(CRUD):
    __tablename__ = "users"

    email: Mapped[str] = mapped_column(String, unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

With the CRUD base, you get immediate access to instance-level operations:

# Create
user = User(email="alice@example.com", name="Alice")
await user.create()

# Read
user = await User.read(user.id)
all_active = await User.read_all(is_active=True)

# Update
await user.update(name="Alice Smith")

# Delete
await user.delete()

Alembic Migration Workflow

Alembic manages schema migrations for the PostgreSQL database. It tracks which migrations have been applied and generates new migration files from model changes.

Initialize Alembic (first time only)
```
cd src && alembic init alembic
```
This creates the alembic/ directory with env.py and a versions/ folder for migration scripts.

Configure alembic/env.py to use the async engine and import your models:

from singularity.db.models import Base
from singularity.db.engine import async_engine

target_metadata = Base.metadata

Generate a migration from model changes:
```
uv run alembic revision --autogenerate -m "add users table"
```
Alembic compares the current database schema against your model definitions and generates an upgrade() / downgrade() migration script.
Apply migrations to bring the database up to date:
```
uv run alembic upgrade head
```
Rollback one migration:
```
uv run alembic downgrade -1
```

Check current state:

uv run alembic current
uv run alembic history --verbose

Always review auto-generated migrations before applying them. Alembic's autogenerate is powerful but may miss certain changes (like column renames, which it interprets as a drop + add). Edit the generated script to use op.alter_column() when needed.

Common Alembic Commands

Command	Description
`alembic revision --autogenerate -m "message"`	Generate migration from model diff
`alembic upgrade head`	Apply all pending migrations
`alembic downgrade -1`	Roll back the last migration
`alembic current`	Show current migration revision
`alembic history --verbose`	List all migrations with details
`alembic stamp head`	Mark the database as up-to-date without running migrations

Script Run Tracker

Singularity includes a BaseScript system for managing database scripts -- seed data, data migrations, cleanup tasks, and other one-time or repeatable operations. These scripts are tracked in a script_run_tracker table.

Table Schema

The script_run_tracker table stores the execution history of every managed script:

Column	Type	Description
`script_name`	`VARCHAR` (PK)	Unique identifier for the script
`status`	`VARCHAR`	Current status: `pending`, `success`, `failed`, `rolled_back`
`error_message`	`TEXT`	Error details if status is `failed`
`content_hash`	`VARCHAR`	SHA-256 hash of the script file for change detection
`executed_at`	`TIMESTAMP`	When the script was first executed
`updated_at`	`TIMESTAMP`	When the record was last modified

How Script Tracking Works

The BaseScript class provides automatic tracking for every script execution:

# singularity/scripts/base_script.py
class BaseScript(ABC):
    def __init__(
        self,
        script_name: str,
        auto_run: bool = False,
        rerun: bool = False,
        deprecated: bool = False,
        affected_tables: Optional[list[str]] = None,
        suppress_errors: bool = False,
    ):
        self.script_name = script_name
        self.auto_run = auto_run
        self.rerun = rerun
        self.deprecated = deprecated
        self._affected_tables = affected_tables
        self.suppress_errors = suppress_errors

    @abstractmethod
    async def execute(self, db: AsyncSession) -> None:
        """Main script logic. Override this."""
        pass

    async def rollback(self, db: AsyncSession) -> None:
        """Optional rollback logic."""
        pass

    async def verify(self, db: AsyncSession) -> bool:
        """Optional post-execution verification. Return True if successful."""
        return True

Hash-Based Change Detection

Every script file is hashed with SHA-256. When a script runs, its hash is stored in script_run_tracker. On subsequent startups, if the file hash has changed, the script is automatically re-executed -- even if it previously succeeded.

Parallel Execution with Table Locks

The ScriptRunner uses asyncio.Lock objects keyed by table name. Scripts that affect different tables run in parallel. Scripts sharing tables are serialized automatically, with locks acquired in alphabetical order to prevent deadlocks.

Script Lifecycle

The execution flow for a managed script follows this sequence:

Discovery -- The ScriptRunner scans scripts/executable/ for classes inheriting from BaseScript.
Status check -- For each script, the runner queries script_run_tracker for previous execution status.
Hash comparison -- The current file hash is compared against the stored content_hash. If they differ, the script is flagged for re-execution.
Lock acquisition -- Table locks are acquired based on affected_tables (inferred from SQL patterns in source code if not explicitly set).
Execution -- execute(db) runs within a database session.
Verification -- verify(db) is called to confirm the script succeeded (defaults to True).
Tracking -- The result is logged to script_run_tracker with the current status and content hash.

Example Script

# scripts/executable/create_default_roles.py
from singularity.scripts.base_script import BaseScript
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text

class CreateDefaultRoles(BaseScript):
    def __init__(self):
        super().__init__(
            "create_default_roles",
            auto_run=True,                    # Run on app startup
            affected_tables=["roles"],        # Lock 'roles' table during execution
        )

    async def execute(self, db: AsyncSession):
        default_roles = ["admin", "editor", "viewer"]
        for role_name in default_roles:
            await db.execute(
                text("INSERT INTO roles (name) VALUES (:name) ON CONFLICT DO NOTHING"),
                {"name": role_name},
            )
        await db.commit()

    async def rollback(self, db: AsyncSession):
        await db.execute(text("DELETE FROM roles WHERE name IN ('admin', 'editor', 'viewer')"))
        await db.commit()

    async def verify(self, db: AsyncSession) -> bool:
        result = await db.execute(text("SELECT COUNT(*) FROM roles"))
        count = result.scalar()
        return count >= 3

Script Management CLI

# List all discovered scripts
singularity scripts list --detailed

# Run a specific script
singularity scripts run create_default_roles

# Force re-run even if already executed
singularity scripts run create_default_roles --force

# Check script status
singularity scripts run create_default_roles status

# Rollback a script
singularity scripts run create_default_roles rollback

# View execution history
singularity scripts history

# Populate missing content hashes (after migration to hash tracking)
singularity scripts sync-hashes

# Run all auto-run scripts (same as app startup)
singularity scripts startup

Using the Database in Services

Services access the database through the Acquire container's db_session attribute, which is the async_sessionmaker factory:

# services/v1/users/service.py
from singularity.core import Service
from sqlalchemy.future import select

class UsersService(Service):
    def __init__(self, acquire):
        super().__init__(acquire)
        self.db = acquire.db_session

    async def get(self):
        async with self.db() as session:
            result = await session.execute(select(User))
            users = result.scalars().all()
            return {"users": [{"id": str(u.id), "email": u.email} for u in users]}

    async def post(self, email: str, name: str):
        async with self.db() as session:
            user = User(email=email, name=name)
            session.add(user)
            await session.commit()
            return {"id": str(user.id), "email": user.email}

Always use async with self.db() as session: to properly scope database sessions. The context manager ensures the session is closed after use, returning the connection to the pool.

Database & Migrations

Hash-Based Change Detection

Parallel Execution with Table Locks

On this page