Implementing the Active Record Pattern in Python with SQLModel

Python developers often miss Active Record’s elegant database interaction when migrating from Ruby on Rails to Python. While Python's SQLAlchemy (and therefore SQLModel) takes a different approach by default, we can implement a similar pattern to bring the convenience of Rails-style models to Python applications while maintaining type safety and following Python best practices.

Active Record Mode

The Active Record pattern (popularized by Ruby on Rails) treats database records as objects with database manipulation methods. There is no need to use a separate repository class or data access object (DAO), the model itself knows how to interact with the database.

For example, in Rails you could write:

<code class="language-ruby"># 查找记录
user = User.find(123)

# 更新记录
user.name = "New Name"
user.save

# 创建新记录
post = Post.create(title: "Hello World")</code>

Using SQLModel in Python

Although Python's SQLModel does not directly provide this mode, we can implement it using a base class that provides these common operations. Here’s how:

1. CRUD-based model

First, we create a base class that implements common CRUD operations:

<code class="language-python">from typing import TypeVar, List, Optional, Tuple
from datetime import datetime
import uuid
from sqlmodel import SQLModel, Session, select
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)).one()

        offset = (page - 1) * per_page
        results = session.exec(
            statement.offset(offset).limit(per_page)
        ).all()

        return results, total</code>

2. Use this pattern in your model

After defining the base class, we can create models that inherit it:

<code class="language-python">class Article(CRUDModel, table=True):
    title: str = Field(..., description="Article title")
    content: str = Field(..., description="Article content")
    status: str = Field(default="draft")

    # Relationships
    comments: List["Comment"] = Relationship(
        back_populates="article",
        sa_relationship_kwargs={"cascade": "all, delete-orphan"}
    )</code>

3. Use Model

Now we can use a Rails-like syntax to consume our models while maintaining Python’s explicit session management:

<code class="language-python">from db.session import 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"Found: {article.title}")

# 创建新文章
with get_session() as session:
    article = Article.create(
        session,
        title="My New Article",
        content="Some content here"
    )

# 更新文章
with get_session() as session:
    article = Article.find(session, "some-uuid")
    if article:
        updated = article.update(
            session,
            title="Updated Title",
            content="New 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)</code>

Main differences with Rails

While this pattern brings Rails-like conveniences to Python, there are some important differences to note:

1. Explicit session management: Python requires explicit session management, which helps to better understand database transactions.

<code class="language-python"># 使用SQLModel的Python
with get_session() as session:
    article = Article.create(session, title="Hello")

# 与Rails对比
article = Article.create(title: "Hello")</code>

2. Type Safety: Python’s type hints provide better IDE support and catch errors earlier.

<code class="language-python">class Article(CRUDModel, table=True):
    title: str  # 类型安全！
    views: int = Field(default=0)</code>

3. Class method: Python uses an explicit @classmethod decorator to handle operations that do not require an instance.
4. Error handling: Python encourages explicit exception handling:

<code class="language-python">with get_session() as session:
    try:
        article = Article.find(session, "non-existent")
        if article is None:
            raise HTTPException(status_code=404, detail="Article not found")
    except Exception as e:
        # 处理其他数据库错误
        raise HTTPException(status_code=500, detail=str(e))</code>

Best Practices

When using this pattern in Python, keep the following best practices in mind:

1. Always use context manager :

<code class="language-python">   # 正确的做法
   with get_session() as session:
       article = Article.create(session, title="Hello")

   # 不正确的方法
   session = get_session()
   article = Article.create(session, title="Hello")
   session.close()</code>

2. Type safety:

<code class="language-python">   # 使用正确的类型提示
   def get_article(id: str) -> Optional[Article]:
       with get_session() as session:
           return Article.find(session, id)</code>

3. Verification:

<code class="language-python">   class Article(CRUDModel, table=True):
       title: str = Field(..., min_length=1, max_length=100)
       status: str = Field(
           default="draft",
           validate_default=True,
           validator=lambda x: x in ["draft", "published"]
       )</code>

4. Relationship Management:

<code class="language-python">   class Article(CRUDModel, table=True):
       # 正确使用级联删除
       comments: List["Comment"] = Relationship(
           back_populates="article",
           sa_relationship_kwargs={"cascade": "all, delete-orphan"}
       )</code>

Conclusion

The Active Record pattern can be efficiently implemented in Python while maintaining type safety and following Python best practices. While it requires more explicit session management than Rails, it offers similar convenience while giving developers more control over database operations.

This mode is especially suitable for:

• The team that migrated from Rails to Python
• Items that prefer model-centric database operations
• Applications where type safety and explicit session management are important

Remember, this is just one way of doing database operations in Python. SQLModel and SQLAlchemy support other modes, such as repositories or data access objects, which may be more suitable for certain use cases.

Resources

• SQLModel Documentation
• FastAPI using SQLModel
• SQLAlchemy Documentation
• Type hints in Python

The above is the detailed content of Implementing the Active Record Pattern in Python with SQLModel. For more information, please follow other related articles on the PHP Chinese website!