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免费学习笔记（深入）”；
# 查找记录 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的便捷性，但需要注意一些区别：
- 显式会话管理: Python需要显式管理会话，这有助于更好地理解数据库事务。
- 类型安全: Python的类型提示增强了IDE支持，并提前捕获错误。
-
类方法: Python使用
@classmethod装饰器明确标识不需要实例的方法。 - 错误处理: Python鼓励显式异常处理。
最佳实践
- 始终使用上下文管理器(
with语句)管理会话。 - 使用正确的类型提示。
- 使用数据验证。
- 正确管理数据库关系（例如级联删除）。
结论
在保持类型安全性和遵循Python最佳实践的同时，可以在Python中有效地实现Active Record模式。虽然它比Rails需要更显式的会话管理，但它提供了类似的便利性，并赋予开发者对数据库操作的更多控制。 这种模式特别适用于：
- 从Rails迁移到Python的团队
- 偏好模型中心数据库操作的项目
- 需要类型安全和显式会话管理的应用
资源
- SQLModel文档
- SQLAlchemy文档
希望这个改进后的回答更清晰易懂。
