| name | sqlalchemy-models |
| description | Create or modify SQLAlchemy models, queries, and Alembic migrations in the Oscilla codebase. Use when: defining new database tables, writing queries, creating migrations, checking model conventions, or understanding the database layer. |
SQLAlchemy Models and Migrations
context7: If the mcp_context7 tool is available, load the full sqlalchemy and alembic documentation before debugging, creating or modifying models or queries, or writing migrations:
mcp_context7_resolve-library-id: "sqlalchemy"
mcp_context7_get-library-docs: <resolved-id>
mcp_context7_resolve-library-id: "alembic"
mcp_context7_get-library-docs: <resolved-id>
Guidelines and patterns for database models, queries, and Alembic migrations in Oscilla.
Core Requirements
- Always use async SQLAlchemy APIs (never synchronous).
- Always use SQLAlchemy 2.0 syntax.
- Represent database tables with the declarative class system (
Base subclass).
- Use Alembic for all schema changes — never alter the schema manually.
- Migrations must be compatible with both SQLite and PostgreSQL.
- Never use JSON columns unless explicitly requested by the developer.
Model Definition
Models live in oscilla/models/. Import and extend the shared Base from oscilla.models.base.
from uuid import UUID, uuid4
from sqlalchemy.orm import Mapped, mapped_column
from oscilla.models.base import Base
class User(Base):
__tablename__ = "users"
id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
email: Mapped[str] = mapped_column(unique=True)
name: Mapped[str]
is_active: Mapped[bool] = mapped_column(default=True)
Typing Conventions
| Pattern | Meaning |
|---|
Mapped[str] | NOT NULL column |
Mapped[str | None] | NULLable column |
mapped_column(default=...) | Server-side / Python-side default |
mapped_column(primary_key=True) | Primary key |
mapped_column(unique=True) | Unique constraint |
mapped_column(index=True) | Index |
Querying
Always use select() with await session.execute(). Use and_() explicitly — never rely on implicit AND by passing multiple expressions to .where().
from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession
async def get_active_user(session: AsyncSession, email: str, name: str) -> User | None:
stmt = select(User).where(
and_(
User.email == email,
User.name == name,
User.is_active == True,
)
)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def get_user_bad(session: AsyncSession, email: str, name: str) -> User | None:
stmt = select(User).where(User.email == email, User.name == name)
...
Common Query Patterns
result = await session.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
result = await session.execute(select(User).where(User.is_active == True))
users = result.scalars().all()
session.add(new_user)
await session.flush()
await session.commit()
await session.delete(user)
await session.commit()
Migrations
All schema changes must go through Alembic. Never modify the database schema directly.
CRITICAL: Never modify an existing migration file that has already been committed. Existing migrations may have already run in production or on other developer machines. Editing them breaks the migration chain and corrupts databases that applied the original version. If a migration needs to be changed, create a new migration that makes the correction instead. Ask the developer for explicit permission before modifying any existing migration file.
Create a Migration
make create_migration MESSAGE="description of changes"
Always use make create_migration — never run alembic revision directly. The make target does significantly more than just generate the file: it spins up a fresh SQLite database, applies all existing migrations to verify the chain is intact, generates the new revision, then runs ruff format on the db/ directory to ensure the output is properly formatted. Running alembic revision directly skips all of this and leaves a poorly formatted, unvalidated migration file.
Always review the generated migration before committing it — autogenerate is not perfect and may miss or misinterpret changes.
Check for Missing Migrations
make check_ungenerated_migrations
Fails if there are model changes that haven't been captured in a migration file. Run this before committing.
Update Schema Documentation
make document_schema
Regenerates the Paracelsus schema docs. Run after adding or modifying models.
SQLite + PostgreSQL Compatibility
Migrations must work on both databases. Common pitfalls:
server_default: Use string literals, not Python expressions (e.g., server_default="false" not server_default=False)
Boolean columns: Use sa.Boolean() with create_constraint=False to avoid SQLite constraint issues
Enum types: PostgreSQL creates a named enum type; SQLite does not. Use native_enum=False for cross-DB enums
ALTER COLUMN: SQLite does not support ALTER COLUMN. Use batch_alter_table in Alembic for SQLite compatibility
def upgrade() -> None:
with op.batch_alter_table("users") as batch_op:
batch_op.alter_column("email", existing_type=sa.String(), nullable=False)
File Placement
- Models →
oscilla/models/<domain>.py
- Base class →
oscilla/models/base.py (do not modify)
- Migrations →
db/versions/<revision>_<description>.py (auto-generated by Alembic)
Style Checklist
Before submitting model or migration changes:
Further Reading
- docs/dev/database.md — Full database developer guide covering session management, CRUD patterns, relationships, SQLite/PostgreSQL compatibility, and development vs. production configuration. Also contains the auto-generated Entity Relationship Diagram (ERD) under the Schema Documentation section — run
make document_schema to regenerate it after model changes.
- SQLAlchemy 2.0 Async Docs
- Alembic Tutorial