用SQLModel在Python中实现主动记录模式-Python教程-PHP中文网

用SQLModel在Python中实现主动记录模式

心靈之曲

发布： 2025-01-27 08:12:34

原创

982人浏览过

用sqlmodel在python中实现主动记录模式

Python开发者在使用SQLModel时，常常怀念Rails中优雅的数据库交互方式。本文将介绍如何在Python中，借助SQLModel实现类似Rails的Active Record模式，兼顾类型安全和Python最佳实践。

许多从Ruby on Rails转到Python的开发者，都会想念Active Record的便捷性。虽然Python的SQLAlchemy（以及SQLModel）采用了不同的方法，但我们可以通过实现一个基类来模拟Active Record的特性，在保持类型安全性的同时，提升Python应用的开发效率。

Active Record模式

Active Record模式将数据库记录视为带有数据库操作方法的对象。它无需单独的Repository类或DAO（数据访问对象），模型本身就具备数据库交互能力。例如，在Rails中：

立即学习“Python免费学习笔记（深入）”；

<code class="ruby"># 查找记录
user = User.find(123)

# 更新记录
user.name = "新名称"
user.save

# 创建新记录
post = Post.create(title: "你好，世界")</code>

登录后复制

使用SQLModel实现Active Record模式

SQLModel本身并不直接支持Active Record模式，但我们可以创建一个基类来实现：

基础CRUD模型

首先，创建一个基类，实现常见的CRUD（创建、读取、更新、删除）操作：

<code class="python">from typing import TypeVar, List, Optional, Tuple
from datetime import datetime
import uuid
from sqlmodel import SQLModel, Session, select, Field
from sqlalchemy import func

T = TypeVar("T", bound="CRUDModel")

class CRUDModel(SQLModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()), primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    @classmethod
    def all(cls: type[T], session: Session) -> List[T]:
        statement = select(cls)
        return session.exec(statement).all()

    @classmethod
    def find(cls: type[T], session: Session, id: str) -> Optional[T]:
        statement = select(cls).where(cls.id == id)
        return session.exec(statement).first()

    @classmethod
    def create(cls: type[T], session: Session, **kwargs) -> T:
        db_obj = cls(**kwargs)
        session.add(db_obj)
        session.commit()
        session.refresh(db_obj)
        return db_obj

    def update(self: T, session: Session, **kwargs) -> T:
        kwargs['updated_at'] = datetime.utcnow()
        for key, value in kwargs.items():
            setattr(self, key, value)
        session.add(self)
        session.commit()
        session.refresh(self)
        return self

    def delete(self: T, session: Session) -> None:
        session.delete(self)
        session.commit()

    @classmethod
    def paginate(
        cls: type[T],
        session: Session,
        page: int = 1,
        per_page: int = 20
    ) -> Tuple[List[T], int]:
        statement = select(cls)
        total = session.exec(select(func.count()).select_from(statement)).scalar()

        offset = (page - 1) * per_page
        results = session.exec(
            statement.offset(offset).limit(per_page)
        ).all()

        return results, total</code>

登录后复制

在模型中使用该模式

通过继承CRUDModel基类，创建新的模型：

如知AI笔记

如知笔记——支持markdown的在线笔记，支持ai智能写作、AI搜索，支持DeepseekR1满血大模型

查看详情

<code class="python">from sqlmodel import Field, Relationship

class Article(CRUDModel, table=True):
    title: str = Field(..., description="文章标题")
    content: str = Field(..., description="文章内容")
    status: str = Field(default="draft")

    # 关系
    comments: List["Comment"] = Relationship(back_populates="article", sa_relationship_kwargs={"cascade": "all, delete-orphan"})

class Comment(CRUDModel, table=True):
    text: str = Field(...)
    article_id: str = Field(foreign_key="article.id")
    article: Article = Relationship(back_populates="comments")</code>

登录后复制

使用模型

现在可以使用类似Rails的语法操作模型，同时保持Python的显式会话管理：

<code class="python">from db.session import get_session  # 假设你已经定义了get_session函数

# 列出所有文章
with get_session() as session:
    articles = Article.all(session)

# 查找特定文章
with get_session() as session:
    article = Article.find(session, "some-uuid")
    if article:
        print(f"找到文章：{article.title}")

# 创建新文章
with get_session() as session:
    article = Article.create(session, title="我的新文章", content="这里是一些内容")

# 更新文章
with get_session() as session:
    article = Article.find(session, "some-uuid")
    if article:
        updated_article = article.update(session, title="更新后的标题", content="新的内容")

# 删除文章
with get_session() as session:
    article = Article.find(session, "some-uuid")
    if article:
        article.delete(session)

# 分页
with get_session() as session:
    articles, total = Article.paginate(session, page=2, per_page=10)</code>

登录后复制

与Rails的主要区别

虽然这种方法带来了Rails的便捷性，但需要注意一些区别：