【Python】使用 Pydantic + SQLAlchemy + MySQL 实现自动记录创建时间和更新时间

news/2024/9/30 8:19:02 标签: python, 开发语言, mysql

使用 Pydantic + SQLAlchemy + MySQL 实现自动记录创建时间和更新时间

在 Web 应用开发中，自动记录数据库中的 创建时间 和 更新时间 是常见的需求。无论是日志记录、数据跟踪，还是审计功能，这类时间戳都至关重要。本文将介绍如何结合 SQLAlchemy 和 Pydantic，在使用 MySQL 作为数据库时，自动处理数据插入和更新时的时间戳。

技术栈

  • MySQL：作为数据库，保存数据记录。
  • SQLAlchemy：用于操作数据库的 ORM 工具，使得开发者可以通过 Python 类来操作数据库表。
  • Pydantic：用于数据验证和序列化，常用于 FastAPI 等 Web 框架。
  • MySQL Connector：通过 MySQL Connector 或 pymysql 作为 MySQL 的驱动。

需求分析

每个表记录需要包含两个时间字段：

  1. 创建时间 (created_at)：记录数据首次插入数据库的时间，只在插入时设置一次。
  2. 更新时间 (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 的数据库。

python">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，可以自动处理这些字段。

python">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 字段。

python">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 等框架时非常有用，能够轻松验证并序列化数据。

python">from pydantic import BaseModel
from datetime import datetime

# 用户创建输入数据模型
class UserCreate(BaseModel):
    name: str
    email: str

# 用户响应输出数据模型
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime
    updated_at: datetime
    
    class Config:
        orm_mode = True

解释

  • UserCreate: 用于验证创建用户时输入的数据，包含 name 和 email。
  • UserResponse: 用于返回给客户端的数据模型，包含 id、name、email 以及 created_at 和 updated_at。
  • orm_mode = True: 允许 Pydantic 从 SQLAlchemy 模型自动生成 Pydantic 模型实例。

步骤 6：实现数据库插入和更新操作

在插入和更新数据时，我们使用 SQLAlchemy 的 Session 来操作数据，同时时间戳字段会自动更新。

python">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.name
        db_user.email = user.email
        db.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() 方法可以自动创建表格：

python"># 创建所有定义的表
Base.metadata.create_all(bind=engine)

之后，可以通过创建 Session 并调用上述的 create_user 和 update_user 函数来操作数据：

python"># 创建数据库 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)

完整代码示例

python">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__ = True
    id = 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: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime
    updated_at: datetime

    class 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.name
        db_user.email = user.email
        db.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 等框架时。


http://www.niftyadmin.cn/n/5684706.html

相关文章

【有啥问啥】In-Context Learning（上下文学习）：深入解析与应用实践

【有啥问啥】In-Context Learning（上下文学习）：深入解析与应用实践

In-Context Learning（上下文学习）：深入解析与应用实践 引言 在人工智能与机器学习的浪潮中，In-Context Learning（上下文学习）作为一种新兴的学习范式，正逐渐崭露头角。与传统的机器学习模型不…
阅读更多...
07_矩形圆形绘制

07_矩形圆形绘制

import cv2 import numpy as np newImageInfo (600,600,3) dst np.zeros(newImageInfo,np.uint8) # 1 2 左上角 3 右下角 4 5 fill -1 >0 line w cv2.rectangle(dst,(150,380),(350,550),(150,200,100),3) # 2 center 3 r cv2.circle(dst,(250,250),(100),(0,0,255),6) …
阅读更多...
1.8 软件业务测试

1.8 软件业务测试

欢迎大家订阅【软件测试】 专栏，开启你的软件测试学习之旅！ 文章目录 前言1 概述2 方法3 测试策略4 案例分析 前言 在软件开发生命周期中，业务测试扮演着至关重要的角色。本文详细讲解了业务测试的定义、目的、方法以及测试策略。 本篇文章参…
阅读更多...
macOS 开发环境配置与应用开发

macOS 开发环境配置与应用开发

✅作者简介：2022年博客新星 第八。热爱国学的Java后端开发者，修心和技术同步精进。 🍎个人主页：Java Fans的博客 🍊个人信条：不迁怒，不贰过。小知识，大智慧。 💞当前专栏…
阅读更多...
WingetUI：可视化Windows常用的命令行包管理工具

WingetUI：可视化Windows常用的命令行包管理工具

推荐一个Windows工具，可以轻松地下载、安装、更新和卸载任何在支持的包管理器上发布的软件。 01 项目简介 WinGetUI 是一个基于 .NET 的开源项目，它旨在提供一个图形用户界面（GUI）来管理包，支持命令比如 Winget、Sco…
阅读更多...
Linux：进程地址空间

Linux：进程地址空间

一、C语言内存管理基础 引入：以前我们知道一个指针指向的如果是一个常量字符串，那么这个就是指向的常量区，只读不可被修改，因此下面的程序会崩溃。 1、在我们C语言内存管理机制里面线性地址是有区域划分的。 我们如何验证这个区域…
阅读更多...
中间件技术

中间件技术

在Java开发中，中间件技术是一种非常关键且广泛使用的技术。中间件通常被定义为位于操作系统、网络和数据库之上的软件层，用于简化分布式系统的开发、部署和管理。它们提供了一系列服务，如消息传递、事务管理、安全控制等，以帮助开…
阅读更多...
【PyTorch】生成对抗网络

【PyTorch】生成对抗网络

生成对抗网络是什么 Generative Adversarial Nets，简称GAN GAN：生成对抗网络 —— 一种可以生成特定分布数据的模型 《Recent Progress on Generative Adversarial Networks (GANs): A Survey》 《How Generative Adversarial Networks and Its Varian…
阅读更多...
最新文章

Copyright @ 2022~2023