Database Management Guide¶

SPOT Platform uses SQLAlchemy for ORM and Alembic for database migrations. This guide covers database operations and best practices.

Quick Start¶

# Check current database revision
make db:status

# Apply pending migrations
make db:migrate

# Show migration history
make db:history

# Create a new migration
make db:revision MESSAGE="Add user table"

Deploy-Time Migrations¶

Important: Migrations must be run at deploy-time, before starting services.

Services do not run migrations automatically at startup. This is a deliberate design choice:

1. Predictable deployments: Migrations run once per deployment, not per service replica
2. Faster startup: Services start immediately without waiting for migration locks
3. No race conditions: Concurrent service startups don't conflict on migrations
4. Clear separation: Infrastructure changes (schema) are separate from application startup

Deployment Workflow¶

# 1. Apply migrations first
make db:migrate

# 2. Then start services
make service:start

CI/CD Integration¶

In CI/CD pipelines, ensure migrations run before service deployment:

deploy:
  script:
    - make db:migrate              # Run migrations first
    - docker compose up -d         # Then start services

Architecture¶

The database layer is organized into focused components:

• Models (shared/src/spot_shared/database/models.py, shared/src/spot_shared/database/knowledge_models.py, shared/src/spot_shared/database/plugin_models.py): Define database schema
• Repositories (shared/src/spot_shared/database/email_repository.py, shared/src/spot_shared/database/repository.py): Store and read emails and analyses, encrypting what they hold of an email
• Session Management (shared/src/spot_shared/database/session.py): Handle database connections
• Migrations (shared/src/spot_shared/database/alembic/): Manage schema changes

Components¶

1. SQLAlchemy Models: Python classes that map to database tables
2. Repositories: The services' way in: they convert between the contract models and the tables
3. Alembic Migrations: Version-controlled schema changes
4. Session Management: Context managers for safe database access

Models vs Contracts¶

SPOT uses two types of models that serve different purposes:

Key Differences:

• Contract Models (spot_sdk): Define what data looks like when transmitted: the email's view sent to analyzers, their verdicts, the workflow's result
• Database Models (core): Define how data is stored in PostgreSQL (tables, relationships, indexes), and what is encrypted

The repositories convert between them:

import os

from spot_shared.database import AnalysisResultRepository, EmailRepository
from spot_shared.database.session import create_async_session_factory

sessions = create_async_session_factory()
emails = EmailRepository(sessions, os.environ["ENCRYPTION_KEY"])
analyses = AnalysisResultRepository(sessions, os.environ["ENCRYPTION_KEY"])

# Intake: the raw message (email_raw) and its EmailView (emails.view_encrypted),
# both encrypted, in one transaction
await emails.store(raw, view, retention_days=365)
view = await emails.get_view(email_id)      # None for an email stored before 2.0
raw = await emails.get_raw(email_id)

# A workflow's OrchestrationResult: the verdict, the score and decided_by in
# clear, the result encrypted, and one analysis_signals row per signal
await analyses.store(result)
result = await analyses.get_result(job_id)

Why separate?

• Contract models: Focused on the plugin contract and the APIs, versioned with the SDK
• Database models: Internal implementation detail, can evolve separately
• No tight coupling: Core can change storage without breaking the contract

What is stored¶

Everything encrypted uses ENCRYPTION_KEY (Fernet). An email past its expires_at is purged with its raw copy, analyses and signals (data_retention_days), and an audit entry older than audit_retention_days after it.

A dataset's member is a copy of an email: an emails row with dataset_id and copied_from, its own raw message, view and label. The purge skips it, and no email list, search, source filter or statistic shows it. Removing it from its dataset, or deleting the dataset, deletes it (ON DELETE CASCADE); deleting the email it was copied from leaves it (Retention).

A workflow dry run stores its message as an email flagged dry_run, for the length of the run, and deletes it when the run ends. No email list, search or source filter shows it; it expires when the gateway stops waiting for the run, so the purge deletes one left behind. A downgrade of migration 013 deletes the dry runs' emails, then the column.

Migration 010 (contract 2) adds email_raw, emails.view_encrypted, and the verdict columns and analysis_signals, and converts the analyses stored in contract 1: is_phishing false is benign; true with a high or critical threat level is malicious, and suspicious otherwise; a failed analysis is unknown. The score is confidence x 100 for a phishing verdict and (1 - confidence) x 100 for a benign one; decided_by is v1; each indicator becomes a signal legacy.<type> of weight 0; what contract 1 recorded besides is kept encrypted (legacy_encrypted). It needs ENCRYPTION_KEY as soon as there are analyses to convert, and stops before changing anything without it. Emails stored before 2.0 keep their contract 1 content and have no raw copy or view.

A downgrade of migration 010 needs ENCRYPTION_KEY too as soon as there are analyses or signals, and stops before changing anything without it. It gives each analysis contract 1's columns back: a converted one recovers what it recorded, a 2.0 one gets a classification derived from its verdict and score, and the signals become indicators. It drops what only 2.0 stores: the workflows' results (result_encrypted), the raw messages (email_raw), the views (view_encrypted) and the senders' addresses (emails.sender_encrypted).

Using the ORM¶

Services use async sessions:

from sqlalchemy import select

from spot_shared.database.models import AnalysisResult
from spot_shared.database.session import get_async_db_session

async with get_async_db_session() as session:
    malicious = await session.scalars(
        select(AnalysisResult.id).where(AnalysisResult.verdict == "malicious")
    )
    # Commit happens automatically on context exit

Synchronous code (migrations, scripts) uses get_db_session() the same way, with with.

Advanced Usage¶

from spot_shared.database.session import create_db_engine, create_session_factory

# Create engine with custom config
engine = create_db_engine(pool_size=20, max_overflow=10)

# Create session factory
SessionFactory = create_session_factory(engine)

# Use in your service
session = SessionFactory()
try:
    # Your database operations
    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

Migration Commands¶

Check Migration Status¶

# Show current database revision
make db:status

# Show migration history (LIMIT=N to cap)
make db:history

Apply Migrations¶

# Apply all pending migrations
make db:migrate

# Or upgrade to a specific revision
make db:migrate REVISION=003_add_priority

Rollback Migrations¶

# Rollback the last migration (REVISION defaults to -1)
make db:downgrade

# Rollback the last 3 migrations
make db:downgrade REVISION=-3

Create New Migration¶

# Autogenerate migration from model changes
make db:revision MESSAGE="Add user roles table"

# Disable autogeneration (for an empty manual migration)
make db:revision MESSAGE="Custom data backfill" AUTOGEN=0

Manual Migration Management¶

For advanced operations not exposed by the Make targets (stamp, branches, merge...), invoke alembic directly inside the api-gateway container:

# Stamp database to specific revision (advanced)
docker compose exec api-gateway alembic -c /app/.venv/lib/python3.11/site-packages/spot_shared/database/alembic.ini stamp 001_initial_schema

Creating Database Models¶

Example Model¶

from sqlalchemy import Column, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid

from spot_shared.database.models import Base

class User(Base):
    """User model following Single Responsibility Principle."""

    __tablename__ = 'users'

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    # Relationships
    emails = relationship("Email", back_populates="user")

Best Practices¶

1. Single Responsibility: Each model represents one table
2. Use UUIDs: For primary keys (better for distributed systems)
3. Add Indexes: For frequently queried columns
4. Use Relationships: Define FK relationships using SQLAlchemy ORM
5. Default Values: Use default= for columns with default values

Migration Workflow¶

1. Modify Models¶

Edit shared/src/spot_shared/database/models.py or shared/src/spot_shared/database/plugin_models.py:

# Add new column
class Email(Base):
    __tablename__ = 'emails'

    # ... existing columns ...

    # New column
    priority = Column(Integer, default=0)

2. Generate Migration¶

make db:revision MESSAGE="Add priority to emails"

This creates a file like alembic/versions/002_add_priority_to_emails.py

3. Review Migration¶

Check the generated file in shared/src/spot_shared/database/alembic/versions/:

def upgrade() -> None:
    op.add_column('emails', sa.Column('priority', sa.Integer(), nullable=True))

def downgrade() -> None:
    op.drop_column('emails', 'priority')

4. Apply Migration¶

make db:migrate

5. Verify¶

make db:status
make db:history

Configuration¶

Database URL¶

Set in .env:

DATABASE_URL=postgresql://spot:spot@postgres:5432/spot

Connection Pooling¶

Configure in your service:

engine = create_db_engine(
    pool_size=10,        # Number of connections to keep open
    max_overflow=20,     # Max connections above pool_size
    pool_recycle=3600,   # Recycle connections after 1 hour
    pool_pre_ping=True   # Verify connections before using
)

Debug Mode¶

Enable SQL logging:

SQL_DEBUG=true

Troubleshooting¶

Migration Failed¶

# Check current state
make db:status

# Review migration history
make db:history

# If stuck, stamp to a specific revision (advanced; runs alembic in
# the api-gateway container)
docker compose exec api-gateway alembic -c /app/.venv/lib/python3.11/site-packages/spot_shared/database/alembic.ini stamp <revision_id>

Connection Issues¶

# Verify DATABASE_URL is set
echo $DATABASE_URL

# Check if postgres is running
docker compose ps postgres

# Test connection
docker compose exec postgres psql -U spot -d spot -c "SELECT 1"

Permission Errors¶

If you see permission errors when creating migrations, make sure the shared/src/spot_shared/database/alembic/versions/ directory is writable.

Directory Structure¶

shared/src/spot_shared/database/
├── alembic/                # Alembic configuration
│   ├── env.py              # Environment configuration
│   ├── script.py.mako      # Migration template
│   └── versions/           # Migration files, 001_initial_schema.py to 017_knowledge_ids_per_source.py
├── alembic.ini             # Alembic config file
├── models.py               # Core SQLAlchemy models
├── knowledge_models.py     # The Knowledge Store's table
├── plugin_models.py        # Plugin-related models
├── email_repository.py     # Emails: raw message and view, encrypted
├── repository.py           # Analyses: verdicts, results and signals
└── session.py              # Session management utilities

Resources¶

• SQLAlchemy Documentation
• Alembic Documentation
• PostgreSQL Documentation