sqlmodel-expert

安装量: 60
排名: #12340

安装

npx skills add https://github.com/bilalmk/todo_correct --skill 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

Core Topics

1. Advanced Model Patterns

See: references/advanced-models.md

  • Relationships: One-to-many, many-to-many, self-referential

  • Inheritance: Single table, joined table, polymorphism

  • Validation: Pydantic validators, custom constraints

  • Mixins: Timestamp, soft delete, reusable patterns

  • Field Types: Enums, JSON, arrays, custom types

  • Indexes: Single, composite, partial indexes

  • Constraints: Unique, check, foreign key cascades

2. Comprehensive Migrations

See: references/migrations.md

  • Alembic Setup: Configuration, env.py for SQLModel

  • Creating Migrations: Autogenerate vs manual

  • Schema Changes: Add/drop columns, rename, change types

  • Data Migrations: Complex data transformations

  • Production Workflow: Zero-downtime migrations

  • Rollback Strategies: Safe downgrade patterns

  • Troubleshooting: Common issues and solutions

3. Query Optimization

See: references/queries-optimization.md

  • N+1 Problem: Solutions with eager loading

  • Query Patterns: Joins, aggregations, subqueries

  • Performance: Indexes, batch operations, profiling

  • Advanced Queries: Window functions, CTEs

  • Bulk Operations: Insert, update, delete at scale

  • Testing: Query counting, explain analyze

Common Patterns

One-to-Many Relationship

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")
class HeroTeamLink(SQLModel, table=True):
    hero_id: int = Field(foreign_key="hero.id", primary_key=True)
    team_id: int = Field(foreign_key="team.id", primary_key=True)
    joined_at: datetime = Field(default_factory=datetime.utcnow)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    teams: List["Team"] = Relationship(
        back_populates="heroes",
        link_model=HeroTeamLink
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List[Hero] = Relationship(
        back_populates="teams",
        link_model=HeroTeamLink
    )

Solving N+1 Query Problem

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

Migration Helper Scripts

Initialize Database

python scripts/init_db.py --url postgresql://user:pass@localhost/db

Migration Operations

./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:

  • User model with timestamp mixin

  • Task model with enums and relationships

  • Team model with many-to-many

  • Tag system with link tables

  • Separate read/write/update models

Copy to your project:

cp assets/example-models.py your-project/app/models.py

Best Practices Checklist

Model Design

Use type hints for all fields Separate read/write/update models 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))

Production Workflow

Development

  • Modify SQLModel models

  • Generate migration: ./scripts/migrate.sh create "description"

  • Review generated migration file

  • Test migration: ./scripts/migrate.sh test

  • Commit migration file

Staging

  • Deploy application code

  • Run migrations: alembic upgrade head

  • Verify data integrity

  • Test application

Production

  • Backup database: pg_dump mydb > backup.sql

  • Deploy in maintenance window

  • Run migrations: alembic upgrade head

  • Monitor logs and metrics

  • Verify application functionality

Zero-Downtime Migration Strategy

For large production databases:

# 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.

返回排行榜