当前位置：首页 > 文章列表 > 文章 > python教程 > 用SQLModel在Python中实现主动记录模式

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

2025-01-27 14:10:11 0浏览收藏

文章小白一枚，正在不断学习积累知识，现将学习到的知识记录一下，也是将我的所得分享给大家！而今天这篇文章《用SQLModel在Python中实现主动记录模式》带大家来了解一下##content_title##，希望对大家的知识积累有所帮助，从而弥补自己的不足，助力实战开发！

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的便捷性，但需要注意一些区别：