使用 Pydantic + SQLAlchemy + MySQL 实现自动记录创建时间和更新时间
在 Web 应用开发中，自动记录数据库中的 创建时间 和 更新时间 是常见的需求。无论是日志记录、数据跟踪，还是审计功能，这类时间戳都至关重要。本文将介绍如何结合 SQLAlchemy 和 Pydantic，在使用 MySQL 作为数据库时，自动处理数据插入和更新时的时间戳。
技术栈
- MySQL：作为数据库，保存数据记录。
- SQLAlchemy：用于操作数据库的 ORM 工具，使得开发者可以通过 Python 类来操作数据库表。
- Pydantic：用于数据验证和序列化，常用于 FastAPI 等 Web 框架。
- MySQL Connector：通过 MySQL Connector 或
pymysql作为 MySQL 的驱动。
需求分析
每个表记录需要包含两个时间字段：
- 创建时间 (
created_at)：记录数据首次插入数据库的时间，只在插入时设置一次。 - 更新时间 (
updated_at)：记录数据最近一次被修改的时间，在每次修改时自动更新。
目标是通过 SQLAlchemy 自动处理这两个字段，并且结合 Pydantic 验证和返回数据。
步骤 1：安装依赖
首先，确保你已经安装了以下必要的库：
pip install sqlalchemy pydantic pymysql mysql-connector-python
SQLAlchemy：用于与数据库交互。Pydantic：用于数据模型验证。PyMySQL或mysql-connector-python：作为 MySQL 的驱动。
步骤 2：配置数据库连接
我们使用 SQLAlchemy 的 create_engine 来连接 MySQL 数据库。假设你已经在 MySQL 中创建了名为 test_db 的数据库。
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# MySQL 数据库连接
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/test_db"# 创建 SQLAlchemy 的 Engine
engine = create_engine(DATABASE_URL)# 创建数据库 session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 基础模型类
Base = declarative_base()
步骤 3：定义包含时间戳的基础模型
接下来我们创建一个基础模型，该模型包含 created_at 和 updated_at 字段。通过 SQLAlchemy 的 Column 和 DateTime，可以自动处理这些字段。
from sqlalchemy import Column, Integer, DateTime, func# 创建带有时间戳的基础模型
class TimestampMixin:created_at = Column(DateTime, default=func.now(), nullable=False)updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), nullable=False)# 用户自定义基础类，所有模型继承自该类
class BaseModel(Base, TimestampMixin):__abstract__ = True # 该模型不被映射到数据库id = Column(Integer, primary_key=True, index=True)
解释
TimestampMixin: 该类包含created_at和updated_at两个字段。created_at使用default=func.now()确保在首次插入数据时自动设置当前时间；updated_at使用onupdate=func.now()确保在每次数据更新时自动更新为当前时间。BaseModel: 作为其他数据模型的基础类，包含id字段以及从TimestampMixin继承的时间戳字段。
步骤 4：定义具体的数据模型
下面我们定义一个用户模型 User，它继承了 BaseModel，并且包含用户的 name 和 email 字段。
from sqlalchemy import Column, String# 用户模型
class User(BaseModel):__tablename__ = 'users'# 用户字段name = Column(String(50), nullable=False)email = Column(String(120), unique=True, nullable=False)
解释
User模型继承自BaseModel，因此它自动拥有id、created_at和updated_at字段。name和email是用户模型的其他字段，分别用于存储用户的名字和邮箱地址。
步骤 5：Pydantic 数据模型
为了确保输入数据的验证和输出数据的格式化，我们使用 Pydantic 创建输入和输出数据模型。Pydantic 在结合 FastAPI 等框架时非常有用，能够轻松验证并序列化数据。
from pydantic import BaseModel
from datetime import datetime# 用户创建输入数据模型
class UserCreate(BaseModel):name: stremail: str# 用户响应输出数据模型
class UserResponse(BaseModel):id: intname: stremail: strcreated_at: datetimeupdated_at: datetimeclass Config:orm_mode = True
解释
UserCreate: 用于验证创建用户时输入的数据，包含name和email。UserResponse: 用于返回给客户端的数据模型，包含id、name、email以及created_at和updated_at。orm_mode = True: 允许 Pydantic 从 SQLAlchemy 模型自动生成 Pydantic 模型实例。
步骤 6：实现数据库插入和更新操作
在插入和更新数据时，我们使用 SQLAlchemy 的 Session 来操作数据，同时时间戳字段会自动更新。
from sqlalchemy.orm import Session# 创建新用户
def create_user(db: Session, user: UserCreate):db_user = User(name=user.name, email=user.email)db.add(db_user)db.commit()db.refresh(db_user) # 获取插入后的最新字段（包括 id 和时间戳）return db_user# 更新用户信息
def update_user(db: Session, user_id: int, user: UserCreate):db_user = db.query(User).filter(User.id == user_id).first()if db_user:db_user.name = user.namedb_user.email = user.emaildb.commit()db.refresh(db_user) # 确保更新时间戳自动更新return db_user
解释
create_user: 创建新的用户，并通过db.commit()提交数据。db.refresh()刷新用户对象，确保返回的对象包含最新的id和时间戳。update_user: 根据user_id查找用户并更新用户信息，db.commit()会自动更新updated_at字段。
步骤 7：创建数据库表并进行操作
在操作数据库之前，我们需要确保数据库表已存在。通过 SQLAlchemy 的 Base.metadata.create_all() 方法可以自动创建表格：
# 创建所有定义的表
Base.metadata.create_all(bind=engine)
之后，可以通过创建 Session 并调用上述的 create_user 和 update_user 函数来操作数据：
# 创建数据库 Session
db = SessionLocal()# 创建用户
new_user = UserCreate(name="Alice", email="alice@example.com")
created_user = create_user(db, new_user)
print(created_user)# 更新用户
updated_user = update_user(db, created_user.id, UserCreate(name="Alice Updated", email="alice.updated@example.com"))
print(updated_user)
完整代码示例
from sqlalchemy import create_engine, Column, Integer, String, DateTime, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from pydantic import BaseModel
from sqlalchemy.orm import Session
from datetime import datetime# MySQL 数据库连接
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/test_db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()# 带有时间戳的基础模型
class TimestampMixin:created_at = Column(DateTime, default=func.now(), nullable=False)updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), nullable=False)# 基础模型类
class BaseModel(Base, TimestampMixin):__abstract__ = Trueid = Column(Integer, primary_key=True, index=True)# 用户模型
class User(BaseModel):__tablename__ = 'users'name = Column(String(50), nullable=False)email = Column(String(120), unique=True, nullable=False)# 创建所有表
Base.metadata.create_all(bind=engine)# Pydantic 输入输出模型
class UserCreate(BaseModel):name: stremail: strclass UserResponse(BaseModel):id: intname: stremail: strcreated_at: datetimeupdated_at: datetimeclass Config:orm_mode = True# 创建新用户
def create_user(db: Session, user: UserCreate):db_user = User(name=user.name, email=user.email)db.add(db_user)db.commit()db.refresh(db_user)return db_user# 更新用户
def update_user(db: Session, user_id: int, user: UserCreate):db_user = db.query(User).filter(User.id == user_id).first()if db_user:db_user.name = user.namedb_user.email = user.emaildb.commit()db.refresh(db_user)return db_user# 使用数据库
db = SessionLocal()
new_user = UserCreate(name="Alice", email="alice@example.com")
created_user = create_user(db, new_user)
print(created_user)updated_user = update_user(db, created_user.id, UserCreate(name="Alice Updated", email="alice.updated@example.com"))
print(updated_user)
总结
通过以上步骤，我们实现了在使用 MySQL 作为数据库时，结合 SQLAlchemy 和 Pydantic 自动记录创建时间和更新时间的功能。SQLAlchemy 负责数据库操作，Pydantic 负责数据验证和序列化，两者结合使得代码更加简洁和安全。在现代 Web 应用开发中，这种模式非常实用，尤其是在使用 FastAPI 等框架时。