# 使用提供的辅助脚本
./scripts/migrate.sh create "add user table"
# 或直接使用 Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head
核心主题
1. 高级模型模式
参见：
🇺🇸English
SQLModel Expert
Advanced SQLModel patterns and comprehensive Alembic migrations for production databases.
Quick Start
Define a Basic Model
from sqlmodel import Field, SQLModel
from typing import Optional
from datetime import datetime
class Task(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str = Field(index=True)
description: Optional[str] = None
completed: bool = Field(default=False)
created_at: datetime = Field(default_factory=datetime.utcnow)
Initialize Database
# Using provided script
python scripts/init_db.py --url postgresql://user:pass@localhost/db
# Or manually
from sqlmodel import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)
Create Migration
# Using provided helper script
./scripts/migrate.sh create "add user table"
# Or directly with Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head
from typing import List
from sqlmodel import Field, Relationship, SQLModel
class Team(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
# One team has many heroes
heroes: List["Hero"] = Relationship(back_populates="team")
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
team_id: Optional[int] = Field(foreign_key="team.id")
# Many heroes belong to one team
team: Optional[Team] = Relationship(back_populates="heroes")
from sqlalchemy.orm import selectinload
# BAD - N+1 queries
users = session.exec(select(User)).all()
for user in users:
posts = user.posts # Each triggers a query!
# GOOD - Eager loading (2 queries total)
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()
for user in users:
posts = user.posts # No additional query!
Creating a Migration
# 1. Modify your model
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
email: str
phone: str # New field added
# 2. Generate migration
# alembic revision --autogenerate -m "add phone to user"
# 3. Review generated migration
def upgrade() -> None:
op.add_column('user', sa.Column('phone', sa.String(), nullable=True))
def downgrade() -> None:
op.drop_column('user', 'phone')
# 4. Apply migration
# alembic upgrade head
./scripts/migrate.sh init # Initialize Alembic
./scripts/migrate.sh create "message" # Create migration
./scripts/migrate.sh upgrade # Apply migrations
./scripts/migrate.sh downgrade # Rollback one
./scripts/migrate.sh current # Show current
./scripts/migrate.sh history # Show history
./scripts/migrate.sh test # Test up & down
Example Models
Use the example models in assets/example-models.py as templates:
Use mixins for common fields (timestamps, soft delete)
Define indexes on foreign keys and frequently queried columns
Use enums for constrained choices
Implement proper validation with Pydantic validators
Relationships
Use back_populates for bidirectional relationships
Create explicit link tables for many-to-many
Consider cascade delete behavior
Use eager loading to prevent N+1 queries
Index foreign key columns
Migrations
Always review autogenerated migrations
One logical change per migration
Test both upgrade and downgrade
Use descriptive migration names
Never edit applied migrations
Add data migrations when changing schemas
Backup database before production migrations
Query Optimization
Use eager loading (selectinload) for relationships
Select only needed columns
Use indexes for WHERE/ORDER BY columns
Batch operations instead of loops
Profile slow queries
Use connection pooling
Troubleshooting Guide
Migration Issues
Problem : Alembic doesn't detect model changes
# Solution: Ensure models are imported in env.py
from app.models import User, Task, Team # Import all models
target_metadata = SQLModel.metadata
Problem : Failed migration
# Check current state
alembic current
# Manually fix issue, then stamp
alembic stamp head
# Or downgrade and retry
alembic downgrade -1
alembic upgrade head
Query Performance
Problem : Slow queries
# Enable query logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
# Use EXPLAIN ANALYZE
explain = session.exec(text("EXPLAIN ANALYZE SELECT ...")).all()
# Profile queries
# See references/queries-optimization.md for detailed patterns
Problem : N+1 queries
# Use selectinload
statement = select(User).options(selectinload(User.posts))
# Or joinedload
from sqlalchemy.orm import joinedload
statement = select(User).options(joinedload(User.posts))
# Phase 1: Add new column (nullable)
def upgrade():
op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))
# Deploy app version that writes to both columns
# Phase 2: Backfill data
def upgrade():
op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")
# Phase 3: Make non-nullable
def upgrade():
op.alter_column('user', 'new_email', nullable=False)
# Deploy app version that reads from new column
# Phase 4: Drop old column
def upgrade():
op.drop_column('user', 'email')
Additional Resources
Advanced Patterns : See references/advanced-models.md for inheritance, polymorphism, composite keys
Migration Guide : See references/migrations.md for Alembic mastery
Query Optimization : See references/queries-optimization.md for performance tuning
This skill provides everything needed for professional SQLModel development and database management.
