npx skills add https://github.com/bilalmk/todo_correct --skill sqlmodel-expert面向生产数据库的高级 SQLModel 模式与全面的 Alembic 迁移。
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)
# 使用提供的脚本
python scripts/init_db.py --url postgresql://user:pass@localhost/db
# 或手动操作
from sqlmodel import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)
# 使用提供的辅助脚本
./scripts/migrate.sh create "add user table"
# 或直接使用 Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head
参见:
Advanced SQLModel patterns and comprehensive Alembic migrations for production databases.
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)
# 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)
# 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
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
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
# 一个团队拥有多个英雄
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")
# 多个英雄属于一个团队
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
)
from sqlalchemy.orm import selectinload
# 错误做法 - N+1 查询
users = session.exec(select(User)).all()
for user in users:
posts = user.posts # 每次都会触发一次查询!
# 正确做法 - 预加载(总共 2 次查询)
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()
for user in users:
posts = user.posts # 没有额外的查询!
# 1. 修改你的模型
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
email: str
phone: str # 新增字段
# 2. 生成迁移
# alembic revision --autogenerate -m "add phone to user"
# 3. 检查生成的迁移
def upgrade() -> None:
op.add_column('user', sa.Column('phone', sa.String(), nullable=True))
def downgrade() -> None:
op.drop_column('user', 'phone')
# 4. 应用迁移
# alembic upgrade head
python scripts/init_db.py --url postgresql://user:pass@localhost/db
./scripts/migrate.sh init # 初始化 Alembic
./scripts/migrate.sh create "message" # 创建迁移
./scripts/migrate.sh upgrade # 应用迁移
./scripts/migrate.sh downgrade # 回滚一次迁移
./scripts/migrate.sh current # 显示当前状态
./scripts/migrate.sh history # 显示历史记录
./scripts/migrate.sh test # 测试升级和降级
使用 assets/example-models.py 中的示例模型作为模板:
复制到你的项目:
cp assets/example-models.py your-project/app/models.py
back_populates问题:Alembic 未检测到模型变更
# 解决方案:确保在 env.py 中导入了所有模型
from app.models import User, Task, Team # 导入所有模型
target_metadata = SQLModel.metadata
问题:迁移失败
# 检查当前状态
alembic current
# 手动修复问题,然后标记
alembic stamp head
# 或者降级并重试
alembic downgrade -1
alembic upgrade head
问题:查询缓慢
# 启用查询日志
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
# 使用 EXPLAIN ANALYZE
explain = session.exec(text("EXPLAIN ANALYZE SELECT ...")).all()
# 分析查询性能
# 详细模式请参见 references/queries-optimization.md
问题:N+1 查询
# 使用 selectinload
statement = select(User).options(selectinload(User.posts))
# 或使用 joinedload
from sqlalchemy.orm import joinedload
statement = select(User).options(joinedload(User.posts))
./scripts/migrate.sh create "description"./scripts/migrate.sh testalembic upgrade headpg_dump mydb > backup.sqlalembic upgrade head适用于大型生产数据库:
# 第一阶段:添加新列(可为空)
def upgrade():
op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))
# 部署同时写入两个列的应用程序版本
# 第二阶段:回填数据
def upgrade():
op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")
# 第三阶段:设置为非空
def upgrade():
op.alter_column('user', 'new_email', nullable=False)
# 部署从新列读取数据的应用程序版本
# 第四阶段:删除旧列
def upgrade():
op.drop_column('user', 'email')
此技能提供了专业 SQLModel 开发和数据库管理所需的一切。
每周安装次数
54
代码仓库
首次出现
2026年1月22日
安全审计
安装于
opencode42
gemini-cli41
codex37
github-copilot35
amp31
kimi-cli30
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
)
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!
# 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
python scripts/init_db.py --url postgresql://user:pass@localhost/db
./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
Use the example models in assets/example-models.py as templates:
Copy to your project:
cp assets/example-models.py your-project/app/models.py
back_populates for bidirectional relationshipsProblem : 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
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))
./scripts/migrate.sh create "description"./scripts/migrate.sh testalembic upgrade headpg_dump mydb > backup.sqlalembic upgrade headFor 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')
This skill provides everything needed for professional SQLModel development and database management.
Weekly Installs
54
Repository
First Seen
Jan 22, 2026
Security Audits
Installed on
opencode42
gemini-cli41
codex37
github-copilot35
amp31
kimi-cli30