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

# 查找记录user = User.find(123)# 更新记录user.name = "新名称"user.save# 创建新记录post = Post.create(title: "你好，世界")

from typing import TypeVar, List, Optional, Tuplefrom datetime import datetimeimport uuidfrom sqlmodel import SQLModel, Session, select, Fieldfrom sqlalchemy import funcT = 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

from sqlmodel import Field, Relationshipclass 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")

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)