Skip to content

4.1 单表模型:mapped_column() 定义主键、字符串、时间戳

在 SQLAlchemy 2.0 中,推荐使用 mapped_column() 来定义模型字段,它比旧版的 Column() 更加类型安全,并且与 Pydantic 的兼容性更好。我们以一个简单的用户表为例,展示如何定义主键、字符串字段和时间戳。

功能名称实例调用方法具体功能、注意事项、必需参数/可选参数
定义主键id: Mapped[int] = mapped_column(primary_key=True)自动递增整数主键,primary_key=True 是必需参数
定义字符串name: Mapped[str] = mapped_column(String(50))字符串类型,String(50) 表示最大长度为 50,必需指定长度
定义时间戳created_at: Mapped[datetime] = mapped_column(default=func.now())默认值为当前时间,func.now() 是 SQLAlchemy 的函数
python
# app/models/user.py
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

# 声明式基类
class Base(DeclarativeBase):
    pass

# 用户模型
class User(Base):
    __tablename__ = "users"
    
    # 主键定义
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # 用户名,最大长度 50
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    
    # 邮箱,最大长度 100
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    
    # 创建时间,默认为当前时间
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    
    # 更新时间,每次更新记录时自动更新
    updated_at: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())

这段代码展示了如何使用 SQLAlchemy 2.0 的新语法定义一个用户模型。每个字段都使用 Mapped[类型] 来声明类型,并通过 mapped_column() 来配置数据库列的属性。注意 created_atupdated_at 的区别:前者只在插入时设置,后者在每次更新时都会刷新。

4.2 外键关联:ForeignKeyrelationship(back_populates=...)

当需要建立表之间的关系时,SQLAlchemy 提供了 ForeignKeyrelationship 来实现。假设我们有一个文章表,每篇文章属于一个用户,我们需要建立一对多的关系。

功能名称实例调用方法具体功能、注意事项、必需参数/可选参数
定义外键user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))指向 users 表的 id 列,必需指定完整的表名和列名
定义关系author: Mapped["User"] = relationship(back_populates="articles")建立双向关系,back_populates 必须与另一端的属性名一致
python
# app/models/article.py
from sqlalchemy import ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from .user import User

class Article(Base):
    __tablename__ = "articles"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # 文章标题
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    
    # 文章内容
    content: Mapped[str] = mapped_column(Text, nullable=False)
    
    # 外键关联到用户表
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    
    # 创建时间
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    
    # 关系定义:指向 User 模型
    author: Mapped["User"] = relationship(back_populates="articles")

# 在 User 模型中添加反向关系
# app/models/user.py (补充)
from typing import List
from sqlalchemy.orm import relationship

class User(Base):
    # ... 其他字段保持不变 ...
    
    # 反向关系:一个用户可以有多篇文章
    articles: Mapped[List["Article"]] = relationship(back_populates="author")

这里的关键是 back_populates 参数,它确保了双向关系的一致性。当你访问 article.author 时,会得到对应的用户对象;当你访问 user.articles 时,会得到该用户的所有文章列表。这种设计让数据查询更加直观和高效。

4.3 索引与约束:在 __table_args__ 中定义唯一索引

为了提高查询性能和保证数据完整性,我们经常需要在数据库表上创建索引和约束。SQLAlchemy 允许我们在模型中通过 __table_args__ 属性来定义这些。

功能名称实例调用方法具体功能、注意事项、必需参数/可选参数
唯一索引__table_args__ = (UniqueConstraint("email"),)确保 email 字段的唯一性,防止重复注册
普通索引Index("idx_username", "username")加速基于 username 的查询
复合索引Index("idx_user_created", "user_id", "created_at")加速复合条件查询
python
# app/models/user.py (更新版本)
from sqlalchemy import String, func, Index, UniqueConstraint
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(100), nullable=False)
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())
    
    # 定义表级约束和索引
    __table_args__ = (
        # 唯一约束:确保邮箱不重复
        UniqueConstraint("email", name="uq_user_email"),
        # 普通索引:加速用户名查询
        Index("idx_user_username", "username"),
        # 复合索引:加速按用户和时间查询
        Index("idx_user_created", "id", "created_at"),
    )

通过 __table_args__,我们可以一次性定义多个约束和索引。这不仅提高了代码的可读性,还确保了数据库结构的一致性。在实际项目中,合理的索引设计可以显著提升查询性能。

4.4 自动时间戳:default=func.now(), onupdate=func.now()

自动管理时间戳是大多数应用的基本需求。SQLAlchemy 提供了简单而强大的机制来处理创建时间和更新时间。

功能名称实例调用方法具体功能、注意事项、必需参数/可选参数
创建时间戳created_at: Mapped[datetime] = mapped_column(default=func.now())插入记录时自动设置为当前时间
更新时间戳updated_at: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())每次更新记录时自动刷新为当前时间
服务器时间func.now()使用数据库服务器的时间,而不是应用服务器的时间
python
# app/models/base.py
from sqlalchemy import func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

class TimestampMixin:
    """时间戳混入类,可被其他模型继承"""
    
    created_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        comment="记录创建时间"
    )
    
    updated_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        onupdate=func.now(),
        comment="记录最后更新时间"
    )

class Base(DeclarativeBase):
    pass

# 使用混入类的用户模型
# app/models/user.py
from .base import Base, TimestampMixin

class User(Base, TimestampMixin):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    
    # 不需要再手动定义 created_at 和 updated_at
    # 它们已经从 TimestampMixin 继承了

使用混入类(Mixin)的方式可以让多个模型共享相同的时间戳逻辑,避免重复代码。这种方式特别适合大型项目,其中很多表都需要类似的自动时间戳功能。注意 func.now() 使用的是数据库服务器的时间,这样可以避免应用服务器和数据库服务器时间不同步的问题。