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

用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中：

# 查找记录
user = user.find(123)

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

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

登录后复制

使用sqlmodel实现active record模式

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

基础crud模型

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

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

登录后复制

在模型中使用该模式

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

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")

登录后复制

使用模型

现在可以使用类似rails的语法操作模型，同时保持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)

登录后复制

与rails的主要区别

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