sqlalchemy-postgres

<essential_principles>

SQLAlchemy 2.0 + Pydantic + PostgreSQL 最佳实践

此技能提供构建可用于生产环境的数据库层的专业指导。

技术栈

• SQLAlchemy 2.0，支持异步（asyncpg 驱动）
• Pydantic v2，用于验证和序列化
• Alembic，用于迁移
• 仅使用 PostgreSQL

核心原则

1. 关注点分离

models/       # SQLAlchemy ORM models (database layer)
schemas/      # Pydantic schemas (API layer)
repositories/ # Data access patterns
services/     # Business logic

2. 类型安全优先 始终使用带有 Mapped[] 类型注解的 SQLAlchemy 2.0 风格：

from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

3. 默认采用异步 在 FastAPI 中使用异步引擎和会话：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("postgresql+asyncpg://...")

4. Pydantic 与 SQLAlchemy 的桥接 保持模型和架构分离，同时确保它们可映射：

# Schema reads from ORM
class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

5. 仓储模式 对数据库操作进行抽象，以便于测试并保持代码整洁。 </essential_principles>

您需要什么帮助？

1. 设置数据库层 - 从头开始初始化 SQLAlchemy + Pydantic + Alembic
2. 定义模型 - 创建带有 Pydantic 架构的 SQLAlchemy 模型
3. 创建迁移 - 生成和管理 Alembic 迁移
4. 查询模式 - 异步 CRUD、联接、预加载、优化
5. 完整实现 - 为某项功能构建完整的数据库层

| 响应 | 工作流 | |----------|----------| | 1, "setup", "initialize", "start" | workflows/setup-database.md | | 2, "model", "define", "create model" | workflows/define-models.md | | 3, "migration", "alembic", "schema change" | workflows/create-migration.md | | 4, "query", "crud", "repository" | workflows/query-patterns.md | | 5, "full", "complete", "feature" | 依次运行 setup → define-models → create-migration |

自动检测触发条件（用户提及以下内容时使用此技能）：

• database、db、sqlalchemy、postgres、postgresql
• model、migration、alembic
• repository、crud、query
• async session、connection pool

<reference_index>

领域知识

参考资料	用途
references/best-practices.md	生产环境模式、安全性、性能
references/patterns.md	仓储、工作单元、常见查询
references/async-patterns.md	异步会话管理、FastAPI 集成
</reference_index>

<workflows_index>

工作流	用途
workflows/setup-database.md	初始化完整数据库层
workflows/define-models.md	创建模型 + 架构 + 关系
workflows/create-migration.md	Alembic 迁移工作流
workflows/query-patterns.md	CRUD 操作和优化
</workflows_index>

<quick_reference>

文件结构

src/
├── db/
│   ├── __init__.py
│   ├── base.py          # DeclarativeBase
│   ├── session.py       # Engine + async session factory
│   └── dependencies.py  # FastAPI dependency
├── models/
│   ├── __init__.py
│   └── user.py          # SQLAlchemy models
├── schemas/
│   ├── __init__.py
│   └── user.py          # Pydantic schemas
├── repositories/
│   ├── __init__.py
│   ├── base.py          # Generic repository
│   └── user.py          # User repository
└── alembic/
    ├── alembic.ini
    ├── env.py
    └── versions/

必要导入

# Models
from sqlalchemy import String, Integer, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase

# Async
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# Pydantic
from pydantic import BaseModel, ConfigDict, Field

连接字符串

# PostgreSQL async
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/dbname"

</quick_reference>

<success_criteria> 满足以下条件时，数据库层即为完整：

• 已配置异步引擎和会话工厂
• 基础模型包含常用字段（id、created_at、updated_at）
• 模型使用 Mapped[] 类型注解
• Pydantic 架构设置了 from_attributes=True
• 已配置异步 Alembic
• 已实现仓储模式
• 已配置用于注入会话的 FastAPI 依赖项
• 已配置适用于生产环境的连接池 </success_criteria>