building-with-sqlalchemy-orm

使用 SQLAlchemy ORM 构建

使用 SQLAlchemy ORM 2.0+、通用 PostgreSQL 模式以及 Neon 特有的无服务器注意事项，构建可用于生产环境的数据库应用。

实施前

收集相关背景信息，确保成功实施：

来源	收集内容
代码库	现有模型、数据库设置、连接模式
对话	学生的具体用例（正在构建什么）、约束条件
技能参考资料	references/ 中的领域模式（API 文档、最佳实践、架构）
用户指南	项目约定、熟练程度

只询问学生自己的需求（领域知识已内置于此技能中）。

---

角色定位

你是一位 Python 数据库架构师，拥有使用 SQLAlchemy ORM 构建生产应用的经验。你既了解通用 PostgreSQL 模式（适用于各种场景），也了解 Neon 特有的无服务器注意事项（自动扩缩容、缩容至零、分支）。你构建过包含多张表的应用，并妥善处理事务、关系和连接池。

---

适用场景

• 根据需求构建数据库模型（将表定义为 Python 类）
• 使用事务安全地实现 CRUD 操作
• 管理表之间的关系（外键、联接）
• 使用筛选、排序和复杂联接查询数据
• 使用适当的配置连接到 PostgreSQL 或 Neon
• 向学习持久化的初学者讲解数据库基础知识

---

核心概念

1. 类即模型（ORM 抽象）

SQLAlchemy 将 Python 类映射到数据库表：

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Expense(Base):
    __tablename__ = 'expenses'

    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    category_id = Column(Integer, ForeignKey('categories.id'))

重要原因：你编写 Python 代码。SQLAlchemy 生成 SQL。你无需手动编写 SQL。

2. 会话即事务（工作单元）

会话将数据库操作组合成一个原子事务：

with Session(engine) as session:
    new_expense = Expense(description="Groceries", amount=45.50, category_id=1)
    session.add(new_expense)
    session.commit()  # All or nothing

重要原因：如果发生任何错误，就不会提交任何操作。这能保证数据库一致性。

3. 关系（将外键用作导航）

在 Python 中定义关系，而不是手动编写联接：

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    category = relationship("Category", back_populates="expenses")

用法：

category = session.query(Category).first()
print(category.expenses)  # All expenses in this category

4. 查询（筛选、排序、联接）

无需编写原始 SQL，即可安全地构造查询：

# Filter: expenses > $50
expensive = session.query(Expense).filter(Expense.amount > 50).all()

# Order: sorted by amount descending
sorted_expenses = session.query(Expense).order_by(Expense.amount.desc()).all()

# Join: expenses with their categories
results = session.query(Expense, Category).join(Category).all()

5. Neon 连接注意事项

Neon 是一种支持自动扩缩容和分支的无服务器 PostgreSQL。主要区别：

• 连接字符串：postgresql+psycopg2://user:pass@host/dbname?sslmode=require
• 始终使用 SSL：?sslmode=require（Neon 强制要求）
• 环境变量：将凭据存储在 .env 中（切勿硬编码）
• 自动暂停：Neon 会在空闲时暂停计算资源，连接池有助于应对此情况

---

决策逻辑

场景	模式	原因
第一个数据库模型	单张表、一种 Column 类型	在建立关系之前，先采用最简单的心智模型
需要关联数据	使用 relationship() + ForeignKey	ORM 会为你处理复杂联接
许多并发请求	使用 pool_size 配置连接池	Neon 会扩展计算资源；连接池能充分利用这些资源
数据一致性至关重要	使用 try/except 处理事务	出错时回滚；保证原子性
希望缩容至零	使用 Neon 无服务器服务 + 配置 echo_pool 的连接池	空闲时自动暂停；收到首个请求时唤醒
调试查询	在引擎中启用 echo=True	查看生成的 SQL

---

工作流：构建预算追踪器

步骤 1：定义模型

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime

Base = declarative_base()

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    date = Column(DateTime, default=datetime.utcnow)
    category_id = Column(Integer, ForeignKey('categories.id'))
    category = relationship("Category", back_populates="expenses")

步骤 2：创建引擎和表

import os
from dotenv import load_dotenv

load_dotenv()

# Connection string from environment
DATABASE_URL = os.getenv("DATABASE_URL")
# Format: postgresql+psycopg2://user:password@host/dbname?sslmode=require

engine = create_engine(DATABASE_URL)
Base.metadata.create_all(engine)

步骤 3：实现 CRUD

def create_expense(session, description, amount, category_id):
    """Create a new expense."""
    try:
        expense = Expense(
            description=description,
            amount=amount,
            category_id=category_id
        )
        session.add(expense)
        session.commit()
        return expense
    except Exception as e:
        session.rollback()
        print(f"Error creating expense: {e}")
        return None

def read_expenses(session, category_id=None):
    """Read expenses, optionally filtered by category."""
    query = session.query(Expense)
    if category_id:
        query = query.filter(Expense.category_id == category_id)
    return query.all()

def update_expense(session, expense_id, amount=None, description=None):
    """Update an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        if amount is not None:
            expense.amount = amount
        if description is not None:
            expense.description = description
        session.commit()
        return expense
    return None

def delete_expense(session, expense_id):
    """Delete an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        session.delete(expense)
        session.commit()
        return True
    return False

步骤 4：通过关系查询

# Get all expenses for a category
category = session.query(Category).filter_by(name="Food").first()
print(category.expenses)  # Uses relationship

# Total spent by category
totals = session.query(
    Category.name,
    func.sum(Expense.amount).label('total')
).join(Expense).group_by(Category.name).all()

for name, total in totals:
    print(f"{name}: ${total:.2f}")

步骤 5：安全地处理事务

def transfer_expense(session, expense_id, new_category_id):
    """Move expense to different category (must succeed fully or not at all)."""
    try:
        expense = session.query(Expense).filter(Expense.id == expense_id).first()
        if not expense:
            raise ValueError(f"Expense {expense_id} not found")

        expense.category_id = new_category_id
        session.commit()
        return True
    except Exception as e:
        session.rollback()
        print(f"Transaction failed, rolled back: {e}")
        return False

步骤 6：连接到 Neon

环境文件（.env）：

DATABASE_URL=postgresql+psycopg2://user:[email protected]/dbname?sslmode=require

使用连接池配置进行连接：

from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # Verify connections before use
    echo=False  # Set to True for debugging
)

---

MCP 集成

将 SQLAlchemy 数据库操作连接到 AI Agent：

# Define an MCP tool that the agent can use
def query_expenses_by_category(category_name: str) -> list:
    """Agent can ask: 'How much did I spend on groceries?'"""
    with Session(engine) as session:
        return session.query(Expense).join(Category).filter(
            Category.name == category_name
        ).all()

def summarize_spending(start_date, end_date) -> dict:
    """Agent can generate reports."""
    with Session(engine) as session:
        return session.query(
            Category.name,
            func.sum(Expense.amount).label('total'),
            func.count(Expense.id).label('count')
        ).join(Expense).filter(
            Expense.date.between(start_date, end_date)
        ).group_by(Category.name).all()

将这些注册为 MCP 工具，以便 Agent（Budget Manager）使用。

---

安全与防护措施

绝不要

• ❌ 在 Python 文件中硬编码凭据
• ❌ 忽略事务的错误处理
• ❌ 不验证就信任用户输入
• ❌ 将机密提交到 git
• ❌ 在生产环境中跳过连接池

务必

• ✅ 使用环境变量存储连接字符串（.env 文件）
• ✅ 使用带回滚操作的 try/except 块封装事务
• ✅ 在执行数据库操作前验证并清理所有用户输入
• ✅ 显式使用 session.commit()（绝不自动提交）
• ✅ 出错时使用 session.rollback()
• ✅ 启用 pool_pre_ping=True 检查连接状态
• ✅ 使用 Neon 时设置 ?sslmode=require（无论如何都会强制启用）

常见错误

错误	影响	修复方法
忘记 session.commit()	更改未保存	始终调用 commit() 或使用上下文管理器
出错时未回滚	数据库中存在部分数据	使用带 rollback() 的 try/except 封装
硬编码凭据	安全漏洞	使用环境变量
未使用连接池	Neon 计算资源扩展效率低下	设置 pool_size 参数
查询中直接使用原始用户输入	SQL 注入	使用参数化查询（ORM 会自动处理）

---

预算跟踪器示例（完整）

请参阅 references/budget-tracker-complete.py，其中包含一个可完整运行的预算跟踪器应用，涵盖：

• 模型定义
• 数据库设置
• CRUD 函数
• 事务处理
• Neon 连接
• 使用示例

---

故障排除

问题	原因	解决方案
ModuleNotFoundError: No module named 'sqlalchemy'	尚未安装	pip install sqlalchemy 或 uv add sqlalchemy
ModuleNotFoundError: No module named 'psycopg2'	缺少 PostgreSQL 驱动程序	pip install psycopg2-binary 或 uv add psycopg2-binary
OperationalError: could not connect to server	连接字符串错误或 Neon 离线	检查 DATABASE_URL 格式，并确认 Neon 项目正在运行
IntegrityError: duplicate key value	插入了重复的唯一字段值	检查该值是否已存在，改用更新操作
ForeignKeyError: could not create foreign key	分类不存在	先创建分类，或使用有效的 category_id
查询速度缓慢	缺少索引或关联关系	查看 references/architecture.md 了解索引模式

---

资源

• 官方文档：https://docs.sqlalchemy.org/en/20/orm/quickstart.html
• Neon 文档：https://neon.com/docs/
• PostgreSQL 类型：https://www.postgresql.org/docs/current/datatype.html