当前位置：首页 > article >正文

【Python】FastAPI框架快速实现后端（一）

article 2025/1/24 13:02:46

FastAPI框架快速实现后端-SQLModel的使用

介绍
正文
- 基础模型
- 模型与表
- 定义数据表模型
- 关系定义

介绍

最近1个多月，用FastAPI做了几个日常工作用的小功能，感觉FastAPI确实很适合这种场景，功能要求简单，交付要求比较急，这个时候一个人就能快速用FastAPI写个东西出来马上就用，在这个背景下，我选择使用FastAPI，并且使用过程中也算是有了些积累，分享一下，本章先说说SQLModel。

FastAPI团队为了进一步降低框架使用门槛，还开发了一个SQLModel的ORM来配套使用，虽然相比sa来说SQLModel限制很多，而且不是很灵活，并发方面也有一些bug，但是确实很简单。

正文

下面的内容默认读者具备一些基本概念，比如模型，ORM，Python

基础模型

一般来说，现在设计数据库的时候都会有一些基础字段，比如主键ID、创建时间、更新时间，SQLModel在这里和其他ORM的使用没什么差别，优先定义BaseModel：

import uuid
from datetime import datetime, timezone
from sqlmodel import SQLModel, Field, DateTime

class BaseModel(SQLModel):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        sa_type=DateTime(timezone=True),
        nullable=False
    )
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        sa_type=DateTime(timezone=True),
        sa_column_kwargs={"onupdate": datetime.now(timezone.utc)}
    )

这里的定义方式指定默认会使用时区，并且updated_at会实现修改自动更新的功能，其他的数据表模型在编写的时候只需要集成BaseModel即可

模型与表

在SQLModel中可以直接通过Table参数来生命该模型是否是数据库表模型，默认是False，如果指定为True，则该模型会与数据库表关联上，比如这是一个普通的模型：

import time
from sqlmodel import SQLModel, Field


class RequestAlertQuery(SQLModel):
    title: str | None = None
    labels: dict | None = Field(description="告警条目的标签json，支持多标签查询")
    start: int | None = Field(default_factory=lambda: int(time.time()-1800))
    end: int | None = Field(default_factory=lambda: int(time.time()))
    status: str | None = None
    level: str | None = None
    limit: int | None = Field(default=5, gt=0, le=100)
    page: int | None = Field(default=1, gt=0)

这是一个关联数据表的模型：

from sqlmodel import Field, Column
from sqlalchemy.dialects.postgresql import TEXT

class SendTemplateModel(BaseModel, table=True):
    """存储告警模板，藉由Jinja2进行模板解析

    Args:
        BaseModel (_type_): _description_
        table (bool, optional): _description_. Defaults to True.
    """
    __tablename__ = "t_alert_send_template"
    name: str = Field(unique=True, max_length=56)
    text: str = Field(sa_column=Column(TEXT, nullable=True), description="Jinja template")
    description: str = Field(nullable=True, description="Description of template", max_length=256)

定义数据表模型

上面我们已经看到了一个简单的数据表模型，其中有一些简单的字段，接下来我们看一个复杂一些的模型：

from pydantic import field_validator

class AlertPushApiPublic(BaseModel):
    name: str = Field(nullable=False, max_length=56)
    type: str = Field(sa_column=Column(Enum(PushType)))

    @field_validator('type', mode='before')
    @classmethod
    def validate_push_type(cls, v):
        if isinstance(v, str) and v in PushType:
            return v
        raise ValueError("Unsupport push type")

class AlertPushApiModel(AlertPushApiPublic, table=True):
    __tablename__ = "t_alert_push_api"
    __table_args__ = (
        UniqueConstraint("name", "type", name="name_type_unique"),
    )
    key: str = Field(unique=True, min_length=32, max_length=32, default_factory=lambda: uuid.uuid4().hex)
    is_open: bool = Field(default=True, nullable=True)
    routers: list["RouterModel"] = Relationship(back_populates="api", cascade_delete=True)
    ws_id: uuid.UUID | None = Field(default=None, foreign_key="t_alert_workspace.id", ondelete="SET NULL")

上面的AlertPushApiPublic模型我作为一个公共的模型，因为有多个模型会继承与它，SQLModel是整合了Pydantic的，所以这里可以直接使用field_validator来进行字段校验，那么这里对其进行了判断，看传入的type字段是否是PushType枚举的一种，如果不是则会返回异常。

__tablename__ 是用来定义表名的，而__table_args__ 则可以提供一些额外的表创建参数，比如这里的唯一约束，我建立了一个name-type的联合唯一约束。

关系定义

上面的模型中有这么一行，这是一个一对多的关系：

    routers: list["RouterModel"] = Relationship(back_populates="api", cascade_delete=True

RouterModel定义如下：


class RouterModel(BaseModel, table=True):
    __tablename__ = "t_alert_router"
    __table_args__ = (
        UniqueConstraint("api_id", "priority", name="rule_priority_unique"),
    )
    priority: int = Field(nullable=False)
    fallthrough: bool = Field(nullable=False)
    api_id: uuid.UUID | None = Field(nullable=True, foreign_key="t_alert_push_api.id", ondelete="CASCADE")
    api: AlertPushApiModel = Relationship(back_populates="routers")
    rules: list["RouterRuleModel"] = Relationship(back_populates="router", cascade_delete=True)
    workspaces: list["WorkspaceModel"] = Relationship(back_populates="routers", link_model=RouterWorkspaceLink)

其中的api字段的定义使其和AlertPushApiModel 完成了关联，需要注意的是RouterModel中还具备多对多的关系，即RouterModel和WorkspaceModel具备多对多关系，通过RouterWorkspaceLink模型完成：

class RouterWorkspaceLink(SQLModel, table=True):
    __tablename__ = "t_router_workspace_link"
    router_id: uuid.UUID = Field(foreign_key="t_alert_router.id", primary_key=True, ondelete="CASCADE")
    workspace_id: uuid.UUID = Field(foreign_key="t_alert_workspace.id", primary_key=True, ondelete="CASCADE")