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

文章小白一枚，正在不断学习积累知识，现将学习到的知识记录一下，也是将我的所得分享给大家！而今天这篇文章《用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模式，但我们可以创建一个基类来实现：

1. 基础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

2. 在模型中使用该模式

通过继承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")

3. 使用模型

现在可以使用类似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的便捷性，但需要注意一些区别：

• 显式会话管理: Python需要显式管理会话，这有助于更好地理解数据库事务。
• 类型安全: Python的类型提示增强了IDE支持，并提前捕获错误。
• 类方法: Python使用@classmethod装饰器明确标识不需要实例的方法。
• 错误处理: Python鼓励显式异常处理。

最佳实践

• 始终使用上下文管理器(with语句)管理会话。
• 使用正确的类型提示。
• 使用数据验证。
• 正确管理数据库关系（例如级联删除）。

结论

在保持类型安全性和遵循Python最佳实践的同时，可以在Python中有效地实现Active Record模式。虽然它比Rails需要更显式的会话管理，但它提供了类似的便利性，并赋予开发者对数据库操作的更多控制。 这种模式特别适用于：

• 从Rails迁移到Python的团队
• 偏好模型中心数据库操作的项目
• 需要类型安全和显式会话管理的应用

资源

• SQLModel文档
• SQLAlchemy文档

希望这个改进后的回答更清晰易懂。

以上就是《用SQLModel在Python中实现主动记录模式》的详细内容，更多关于的资料请关注golang学习网公众号！