Schema Management¶
Ferro integrates with Alembic, the industry-standard migration tool for Python, to provide robust and reliable schema management for production environments.
Why Alembic?¶
Instead of reinventing migrations, Ferro leverages Alembic's battle-tested workflow. Ferro provides a bridge that translates your models into SQLAlchemy metadata, which Alembic uses to detect schema changes.
Installation¶
Install Ferro with Alembic support:
This installs Alembic and SQLAlchemy (used only for migration generation, not at runtime).
Quick Start¶
1. Initialize Alembic¶
In your project root:
This creates:
2. Configure env.py¶
Edit migrations/env.py to connect Ferro models to Alembic:
# migrations/env.py
from ferro.migrations import get_metadata
# Import all models to register them
from myapp.models import User, Post, Comment
# Ferro generates SQLAlchemy metadata from registered models
target_metadata = get_metadata()
# Rest of env.py remains unchanged
Column nullability (get_metadata())¶
Ferro maps each model field to a SQLAlchemy Column with a nullable flag used by autogenerate.
- With the default
nullable="infer"onFerroField,ForeignKey, andferro.Field(...), the column is nullable if and only if the field’s Python annotation allowsNone(for exampleT | None). Having a default ordefault_factorydoes not by itself make a column nullable in the migration metadata. - Shadow
{name}_idforeign-key columns infer from the forward relation field’s annotation, not from the synthetic*_idfield (which often uses| Nonefor assignment convenience). ForeignKey(..., on_delete="SET NULL")implies a nullable shadow FK column unless you explicitly override it;nullable=Falseis rejected for that combination.- Set
nullable=Falseornullable=TrueonFerroField,ForeignKey, orferro.Field(...)to force NOT NULL or NULL when you intentionally diverge from the type (for exampleint | Nonefor a type checker while keeping a NOT NULL column).
Primary keys are always emitted as nullable=False.
3. Generate Your First Migration¶
Alembic compares your models to the database and generates a migration script in migrations/versions/.
Composite uniques and Alembic¶
When you declare __ferro_composite_uniques__ on a model, Ferro’s get_metadata() bridge adds matching SQLAlchemy UniqueConstraint objects to the reflected Table. Autogenerated revisions will therefore include those constraints (and the same for default many-to-many join tables, which get a composite unique on the two FK columns). Review generated migrations as usual before applying them in production.
The same applies to __ferro_composite_indexes__: autogen emits matching non-unique sa.Index objects, including the default reverse-direction index on M2M join tables (opt out per-relation with ManyToMany(reverse_index=False)).
For a one-to-one forward relation, declare ForeignKey(..., unique=True) on the annotated field. Ferro’s get_metadata() maps that to unique=True on the shadow foreign-key column so autogenerate emits the same UNIQUE constraint as connect(..., auto_migrate=True) does at runtime.
4. Review the Migration¶
Open the generated file in migrations/versions/xxxx_initial_schema.py:
def upgrade():
op.create_table('users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('username'),
sa.UniqueConstraint('email')
)
# ... more tables
def downgrade():
op.drop_table('users')
# ... reverse operations
Always review generated migrations for correctness.
5. Apply the Migration¶
Your database now matches your models.
Workflow¶
The typical development workflow:
- Modify models in Python
- Generate migration:
alembic revision --autogenerate -m "Description" - Review migration in
migrations/versions/ - Apply migration:
alembic upgrade head - Commit migration to version control
Common Operations¶
Check Current Version¶
View Migration History¶
Upgrade to Specific Version¶
alembic upgrade <revision>
# Examples
alembic upgrade +1 # Upgrade one version
alembic upgrade abc123 # Upgrade to specific revision
alembic upgrade head # Upgrade to latest
Downgrade (Rollback)¶
alembic downgrade -1 # Downgrade one version
alembic downgrade abc123 # Downgrade to specific revision
alembic downgrade base # Downgrade to empty database
Create Empty Migration¶
For custom SQL or data migrations:
Edit the generated file:
def upgrade():
# Custom SQL
op.execute("""
INSERT INTO users (username, email, role)
VALUES ('admin', 'admin@example.com', 'admin')
""")
def downgrade():
op.execute("DELETE FROM users WHERE username = 'admin'")
Production Workflow¶
Development¶
- Develop features with models
- Generate migrations
- Test migrations locally
- Commit migrations to git
Staging¶
- Pull latest code
- Run
alembic upgrade head - Test application
Production¶
- Backup database before migrations
- Review migration scripts
- Run migrations:
- Monitor application
Rollback Strategy¶
Keep rollback migrations tested:
# Test upgrade
alembic upgrade head
# Test downgrade
alembic downgrade -1
# Upgrade again
alembic upgrade head
Precision Mapping¶
Ferro's migration bridge ensures high fidelity between your models and the database:
Nullability¶
# Required field
username: str
# → NOT NULL column
# Optional field
bio: str | None = None
# → NULL allowed
Complex Types¶
from decimal import Decimal
from datetime import datetime
from uuid import UUID
from enum import Enum, StrEnum
class UserRole(StrEnum):
USER = "user"
ADMIN = "admin"
class User(Model):
# Maps to DECIMAL/NUMERIC
balance: Decimal
# Maps to TIMESTAMP
created_at: datetime
# Maps to UUID (or TEXT in SQLite)
id: UUID
# Maps to a named ENUM on PostgreSQL (or VARCHAR + CHECK on SQLite)
role: UserRole
# Maps to JSON/JSONB
metadata: dict
Enums and PostgreSQL: get_metadata() maps Python Enum / StrEnum fields to
sqlalchemy.Enum with an explicit type name derived from the enum class name
(lowercased), for example UserRole → userrole. Autogenerated Alembic
revisions can then compile on PostgreSQL, which rejects anonymous enum types.
Integer-valued enums use string labels in the database type (for example
"1", "2") so the column remains a string-backed enum.
Constraints¶
from ferro import Field, Model
class Product(Model):
# PRIMARY KEY
id: int | None = Field(default=None, primary_key=True)
# UNIQUE constraint
sku: str = Field(unique=True)
# INDEX
category: str = Field(index=True)
Foreign Keys¶
class Post(Model):
author: Annotated[User, ForeignKey(related_name="posts")]
# → FOREIGN KEY (author_id) REFERENCES users(id)
# With cascade
author: Annotated[User, ForeignKey(
related_name="posts",
on_delete="CASCADE"
)]
# → FOREIGN KEY ... ON DELETE CASCADE
Many-to-Many¶
class Student(Model):
courses: Relation[list["Course"]] = ManyToMany(related_name="students")
# Automatically generates join table:
# CREATE TABLE student_courses (
# student_id INT REFERENCES students(id),
# course_id INT REFERENCES courses(id),
# PRIMARY KEY (student_id, course_id)
# )
Data Migrations¶
For migrations that modify data (not just schema):
from alembic import op
import sqlalchemy as sa
def upgrade():
# Schema change
op.add_column('users', sa.Column('role', sa.String(), nullable=True))
# Data migration
connection = op.get_bind()
connection.execute(
"UPDATE users SET role = 'user' WHERE role IS NULL"
)
# Make non-nullable after populating
op.alter_column('users', 'role', nullable=False)
def downgrade():
op.drop_column('users', 'role')
Zero-Downtime Migrations¶
For production systems that can't tolerate downtime:
1. Additive Changes First¶
# Step 1: Add new column (nullable)
def upgrade():
op.add_column('users', sa.Column('new_email', sa.String(), nullable=True))
# Deploy application that writes to both old and new columns
# Wait for all instances to deploy
# Step 2: Migrate data
def upgrade():
connection = op.get_bind()
connection.execute("UPDATE users SET new_email = email WHERE new_email IS NULL")
# Step 3: Make non-nullable, drop old column
def upgrade():
op.alter_column('users', 'new_email', nullable=False)
op.drop_column('users', 'email')
op.alter_column('users', 'new_email', new_column_name='email')
2. Feature Flags¶
Use feature flags to control when code uses new schema:
Troubleshooting¶
PostgreSQL ENUM type requires a name¶
If you see CompileError: PostgreSQL ENUM type requires a name when applying a
migration, upgrade Ferro so get_metadata() emits named enum types, then
regenerate the revision. Older migrations can be fixed by adding name=... to
each sa.Enum(...) call (often matching the Python enum class name in
lowercase).
Migration Not Detected¶
# Ensure models are imported in env.py
from myapp.models import * # Import all models
# Verify metadata generation
target_metadata = get_metadata()
print(target_metadata.tables) # Should list your tables
Conflicting Migrations¶
# Error: Multiple head revisions
# Solution: Merge migrations
alembic merge heads -m "Merge migrations"
Manual Schema Changes¶
Reset Migrations¶
# Delete all migration files
rm migrations/versions/*.py
# Drop all tables
# Then regenerate from scratch
alembic revision --autogenerate -m "Initial schema"
alembic upgrade head
Best Practices¶
- Always review generated migrations
- Test migrations locally before production
- Backup database before running migrations
- Keep migrations small and focused
- Don't edit applied migrations (create new ones)
- Version control all migration files
- Test rollback (downgrade) functionality
- Use descriptive names for migrations
See Also¶
- Database Setup - Connection configuration
- Models & Fields - Model definitions
- How-To: Testing - Testing with migrations