FastAPI与SQLModel

是时候把他们放在一起了

FastAPI和SQLModel的相性非常好（毕竟是一个作者），能轻松的结合在一起提供服务

从需求出发……

如果我们现在有一类资源，我们想对外部提供API来访问这一类资源。对于典型的RESTful API，我们需要实现的接口有：

GET：获取一项/多项资源
POST：新建一个资源
PUT：更新一个资源，提供完整资源数据
PATCH：更新一个资源，但是只需要提供更改的部分
DELETE：删除一个资源

由于PATCH实际上能做所有PUT能做的事情，我们暂时先不考虑PUT。又考虑到GET有两种调用方法，那我们实际上有五个API需要实现，这五个API一般是（以Hero模型为例）：

GET /heroes/：获取多个Hero
POST /heroes/：创建一个hero
GET /heroes/{hero_id}：获取id为hero_id的Hero
PATCH /heroes/{hero_id}：更新id为hero_id的Hero
DELETE /heroes/{hero_id}：删除id为hero_id的Hero

那么接下来，我们就来逐一实现这些API

创建一个资源

我们从创建资源开始。还是以Hero类为例，我们的Hero长成这样：

class Hero(SQLModel, table=True): 
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

如果你对数据库熟悉，你应该会发现，此处的id字段在数据库中应该是一个自增的变量，在创建资源时理应不需要手动指定。虽然此处的id是int | None，但这仍留给了我们手动指定id的可能性，我们可不想这样。因此，在创建Hero资源时，应不允许填写id字段

为了明确这一需求，我们需要将创建Hero的类从普通的Hero类中剥离出来。我们可以这么做：

# 嘘...先别急着真的这么做
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

class HeroCreate(SQLModel):
    # 这里不需要tabel=True, 因为这张表不需要真的存在数据库里
    name: str
    secret_name: str
    age: int | None = None

但是显然，这样编写会存在很多重复字段，非常折磨

为了避免这种问题，我们可以将共同字段提取成一个基类，而Hero和HeroCreate继承这个基类：

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: int | None = Field(defualt=None, primary_key=True)

class HeroCreate(HeroBase):
    pass
    """
    Oh, 真奇怪
    你肯定想问，这个类既然什么都没有定义，它为什么要存在？
    其实这个类是给人看的，因为API中使用“HeroBase”作为类名会令人迷惑
    因此我们创建一个新类，即使它什么都没定义
    """

一般来说，在创建完资源后，我们会返回这个资源的信息，因此我们也同样需要一个类来返回资源信息。在返回时，我们希望携带该资源的id

class HeroPublic(HeroBase):
    id: int

直到这时，我们才开始编写API

我们可以这么来写：

# 省略部分代码，假设你已经完成了engine的创建

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero

出现了一些新东西，让我们来进行说明：

response_model对API返回的数据进行验证和过滤，它能将我们返回的db_hero从Hero类转换到HeroPublic类，同时自动在文档中记录我们的返回格式
hero: HeroCreate代表了我们需求的参数hero是一个HeroCreate类，这表明我们需在请求的Body中给出HeroCreate的所有字段（也就是name，secret_name和age）。这也会在文档中体现

hero原本是一个HeroCreate类型的对象，我们通过Hero.model_validate()将其验证并转换为Hero类

这样，我们就完成了我们的第一个API。呼——

删改查

查

查询分为查询所有资源和查询单个资源，但他们都很简单！

from fastapi import FastAPI, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select

# 省略部分代码...

# 查询所有资源
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = sesson.exec(select(Hero)).all()
        return heroes

# 查询单个资源
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:  # 检验hero是否存在
            raise HTTPException(status_code=404, detail="Hero not found")
        return hero

这两个API的代码逻辑都很好理解。唯一要注意的是，在读取单个Hero时，你还额外需要校验一下用户输入的hero_id是否有对应的Hero存在

对于读取多个资源的API，我们暂时不考虑分页的需求。也许在之后我们会再提到这件事

改

要更新数据，过程比查找会稍微复杂一些

在上文的演示中，我们为了满足Hero在不同API下的不同字段要求，构造了几个不同的Hero相关类。而对于改，我们也有新的需求：我们希望不能改动id字段，而其他字段都可以选择改动

为了满足这样的需求，我们设计HeroUpdate类如下：

class HeroUpdate(SQLModel):
    name: str | None = None
    secret_name: str | None = None
    age: int | None = None

你可能注意到了，我们并没有继承HeroBase类，因为我们的字段要求和他有不小区别。我们所有字段都是可选的，这代表你在更新Hero信息时，可以仅输入其中部分字段进行更新，而非必须更新全部字段

接下来，让我们编写改的API：

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
    with Session(engine) as session:
        db_hero = session.get(Hero, hero_id)
        if not db_hero:
            raise HTTPException(status_code=404, detail="Hero not found")
        hero_data = hero.model_dump(exclude_unset=True)
        db_hero.sqlmodel_update(hero_data)

        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)

        return db_hero

似乎细节有点多…照例，让我们逐一讲解一下：

函数声明中的hero: HeroUpdate代表了要接受HeroUpdate类所定义的数据。HeroUpdate是SQLModel的子类，SQLModel是Pydantic的子类，而Pydantic在FastAPI的参数中会被解释为需要在请求的Body填写的字段。因此，API请求的Body中可以（选择性地）带上name，secret_name，age字段（因为我们将其设为了可选的）
model_dump()方法能将该模型实例转换为字典，可能类似于：

{
    "name": None,
    "secret_name": None,
    "age": 11
}

若model_dump()的exclude_unset参数被设为True（就像上方代码演示的那样），那其在把模型转化为字典时会去除所有请求中未声明的字段（例如在上方就只保留age）
并非只要值是None的字段就一定会被删除，实际上，如果字段在数据库中的值不为None，而客户端想要删除这个字段的值，那么只需要发送这个字段为None的请求即可
在此之后，sqlmodel_update()方法会用那个导出的字典来更新自己模型的数据

如果你参考了一些比较老的代码，你会发现取代sqlmodel_update()的是手动获取hero_data的key-value对，然后使用setattr()方法手动添加（是的我也这么做过）

这是旧版本的做法。新版本添加了sqlmodel_update()方法，你就不需要这么做了

好时代，来临力

删

这个超简单

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int):
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            raise HTTPException(status_code=404, detail="Hero not found")
        session.delete(hero)
        session.commit()
        return {"ok": True}

是的，我们没什么可以返回的东西了，所以我们返回{"ok": True}

总结，以及下一步

到这里为止，你已经可以初步的把FastAPI和SQLModel结合在一起使用了！恭喜！

但是，如果你不想让你的前端朋友太糟心，你一定会需要写一些更复杂的交互逻辑，例如鉴权、关系模型、依赖注入等等。现在这点知识是肯定不够的

对于这些信息，你当然可以查看FastAPI文档，也可以查看SQLModel文档（虽然截至2024.11还没有官方汉化，只有比较烂的疑似机翻），这些都能帮助你达成你的目的

如果有机会的话，我也会再写几篇文章，来介绍一下与FastAPI相关的这些技术

从需求出发……

创建一个资源

删改查

查

改

删

总结，以及下一步

发送评论 编辑评论

推荐文章

发送评论编辑评论