FastAPI入门：中间件、CORS跨域资源共享、SQL数据库

中间件

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

要创建中间件你可以在函数的顶部使用装饰器 @app.middleware(“http”)

中间件参数接收如下参数:

• request.
• 一个函数 call_next 它将接收 request 作为参数.
  • 这个函数将 request 传递给相应的 路径操作.
  • 然后它将返回由相应的路径操作生成的 response.
• 然后你可以在返回 response 前进一步修改它.

在任何路径操作收到request前,可以添加要和请求一起运行的代码.

也可以在响应生成但是返回之前添加代码.

from fastapi import FastAPI, requests
from h11 import Request
import time

app = FastAPI()

@app.middleware("http") # 声明是用于处理HTTP的中间件
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter() # Python 中用于高精度时间测量的函数，专门用于测量代码执行时间
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORS跨域资源共享

CORS 或者「跨域资源共享」 指浏览器中运行的前端拥有与后端通信的 JavaScript 代码，而后端处于与前端不同的「源」的情况。

源：源是协议（http，https）、域（myapp.com，localhost，localhost.tiangolo.com）以及端口（80、443、8080）的组合

假设你的浏览器中有一个前端运行在 http://localhost:8080，并且它的 JavaScript 正在尝试与运行在 http://localhost 的后端通信（因为我们没有指定端口，浏览器会采用默认的端口 80）。

然后，浏览器会向后端发送一个 HTTP OPTIONS 请求，如果后端发送适当的 headers 来授权来自这个不同源（http://localhost:8080）的通信，浏览器将允许前端的 JavaScript 向后端发送请求。

为此，后端必须有一个「允许的源」列表。

在这种情况下，它必须包含 http://localhost:8080，前端才能正常工作

使用 CORSMiddleware
你可以在 FastAPI 应用中使用 CORSMiddleware 来配置它

导入 CORSMiddleware。

• 创建一个允许的源列表（由字符串组成）。
• 将其作为「中间件」添加到你的 FastAPI 应用中。

你也可以指定后端是否允许：

• 凭证（授权 headers，Cookies 等）。
• 特定的 HTTP 方法（POST，PUT）或者使用通配符 “*” 允许所有方法。
• 特定的 HTTP headers 或者使用通配符 “*” 允许所有 headers。

from pydoc import allmethods
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

支持以下参数：

• allow_origins - 一个允许跨域请求的源列表。例如 [‘https://example.org’, ‘https://www.example.org’]。你可以使用 [‘*’] 允许任何源。
• allow_origin_regex - 一个正则表达式字符串，匹配的源允许跨域请求。例如 ‘https://.*.example.org’。
• allow_methods - 一个允许跨域请求的 HTTP 方法列表。默认为 [‘GET’]。你可以使用 [‘*’] 来允许所有标准方法。
• allow_headers - 一个允许跨域请求的 HTTP 请求头列表。默认为 []。你可以使用 [‘*’] 允许所有的请求头。Accept、Accept-Language、Content-Language 以及 Content-Type .请求头总是允许 CORS 请求。
• allow_credentials - 指示跨域请求支持 cookies。默认是 False。另外，允许凭证时 allow_origins 不能设定为 [‘*’]，必须指定源。
• expose_headers - 指示可以被浏览器访问的响应头。默认为 []。
• max_age - 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600。

SQL数据库

SQLModel 是基于 SQLAlchemy 和 Pydantic 构建的。它由 FastAPI 的同一作者制作，旨在完美匹配需要使用 SQL 数据库的 FastAPI 应用程序

由于 SQLModel 基于 SQLAlchemy，因此您可以轻松使用任何由 SQLAlchemy 支持的数据库（这也让它们被 SQLModel 支持），例如：

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server 等.

在这个例子中，我们将使用 SQLite

首先，确保您创建并激活了虚拟环境，然后安装了 sqlmodel

pip install sqlmodel

创建含有单一模型的应用程序

导入 SQLModel 并创建一个数据库模型

from turtle import st
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlalchemy import table
from sqlmodel import Field, Session, SQLModel, create_engine, select

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

Hero 类与 Pydantic 模型非常相似（实际上，从底层来看，它确实就是一个 Pydantic 模型）。

• table=True 会告诉 SQLModel 这是一个表模型，它应该表示 SQL 数据库中的一个表，而不仅仅是一个数据模型（就像其他常规的 Pydantic 类一样）。

• Field(primary_key=True) 会告诉 SQLModel id 是 SQL 数据库中的主键（您可以在 SQLModel 文档中了解更多关于 SQL 主键的信息）。

• 把类型设置为 int | None ，SQLModel 就能知道该列在 SQL 数据库中应该是 INTEGER 类型，并且应该是 NULLABLE 。

• Field(index=True) 会告诉 SQLModel 应该为此列创建一个 SQL 索引，这样在读取按此列过滤的数据时，程序能在数据库中进行更快的查找。

• SQLModel 会知道声明为 str 的内容将是类型为 TEXT （或 VARCHAR ，具体取决于数据库）的 SQL 列。

创建引擎（Engine）
SQLModel 的引擎 engine（实际上它是一个 SQLAlchemy engine ）是用来与数据库保持连接的。

您只需构建一个 engine，来让您的所有代码连接到同一个数据库。

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

使用 check_same_thread=False 可以让 FastAPI 在不同线程中使用同一个 SQLite 数据库。这很有必要，因为单个请求可能会使用多个线程（例如在依赖项中）

创建表
添加一个函数，使用 SQLModel.metadata.create_all(engine) 为所有表模型创建表。

def create_db_and_tables():
    #扫描所有继承自 SQLModel 且设置了 table=True 的类
    # 根据这些类的定义在数据库中创建对应的表
    SQLModel.metadata.create_all(engine)

创建会话（Session）依赖项

Session 会存储内存中的对象并跟踪数据中所需更改的内容，然后它使用 engine 与数据库进行通信。

    
def get_session():
    # Session是一个数据库会话类，代表一个工作单元
    with Session(engine) as session:
        yield session
        
SessionDep = Annotated[Session, Depends(get_session)]

我们会使用 yield 创建一个 FastAPI 依赖项，为每个请求提供一个新的 Session 。这确保我们每个请求使用一个单独的会话

在启动时创建数据库表

@app.on_event("startup")
def on_startup():
    # 在应用启动时创建数据库和表
    create_db_and_tables()

在应用程序启动事件中，我们创建了表

创建 Hero 类
因为每个 SQLModel 模型同时也是一个 Pydantic 模型，所以您可以在与 Pydantic 模型相同的类型注释中使用它。

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero) # 将 hero 对象添加到当前数据库会话中
    session.commit()
    session.refresh(hero) # 刷新 hero 对象，确保获取最新的数据库状态
    return hero

这里，我们使用 SessionDep 依赖项（一个 Session ）将新的 Hero 添加到 Session 实例中，提交更改到数据库，刷新 hero 中的数据，并返回它。

读取 Hero 类

可以使用 select() 从数据库中读取 Hero 类，并利用 limit 和 offset 来对结果进行分页

@app.get("/heroes/")
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

读取单个 Hero

@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    hero = session.get(Hero, hero_id) # get(要查询的模型类，主键值)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

删除单个 Hero

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    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}

运行程序，打开docs文档





更新应用程序以支持多个模型

如果您查看之前的应用程序，您可以在 UI 界面中看到，到目前为止，由客户端决定要创建的 Hero 的 id 值。

我们不应该允许这样做，因为他们可能会覆盖我们在数据库中已经分配的 id 。决定 id 的行为应该由后端或数据库来完成，而非客户端

此外，我们为 hero 创建了一个 secret_name ，但到目前为止，我们在各处都返回了它，使安全性下降

我们将通过添加一些额外的模型来解决这些问题

创建多个模型

有了 SQLModel，我们就可以利用继承来在所有情况下避免重复所有字段。

我们从一个 HeroBase 模型开始，该模型具有所有模型共享的字段：

• name
• age

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

接下来，我们创建 Hero ，实际的表模型，并添加那些不总是在其他模型中的额外字段：

• id
• secret_name

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

接下来，我们创建一个 HeroPublic 模型，这是将返回给 API 客户端的模型。

它包含与 HeroBase 相同的字段，因此不会包括 secret_name , 其中 id 声明为 int （不是 None ）

    
class HeroPublic(HeroBase):
    id: int

现在我们创建一个 HeroCreate 模型，这是用于验证客户数据的模型。

它不仅拥有与 HeroBase 相同的字段，还有 secret_name 。

现在，当客户端创建一个新的 hero 时，他们会发送 secret_name ，它会被存储到数据库中，但这些 secret_name 不会通过 API 返回给客户端

class HeroCreate(HeroBase):
    secret_name: str

HeroUpdate - 用于更新 hero 的数据模型.
它包含创建新 hero 所需的所有相同字段，但所有字段都是可选的（它们都有默认值）。这样，当您更新一个 hero 时，您可以只发送您想要更新的字段

使用 HeroCreate 创建并返回 HeroPublic

我们在请求中接收到一个 HeroCreate 数据模型，然后从中创建一个 Hero 表模型。

这个新的表模型 Hero 会包含客户端发送的字段，以及一个由数据库生成的 id 。

然后我们将与函数中相同的表模型 Hero 原样返回。但是由于我们使用 HeroPublic 数据模型声明了 response_model ，FastAPI 会使用 HeroPublic 来验证和序列化数据

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep) -> Hero:
    db_hero = Hero.model_validate(hero) # 将 HeroCreate 模型转换为 Hero 模型
    session.add(db_hero) # 将 db_hero 对象添加到当前数据库会话中
    session.commit()
    session.refresh(db_hero) # 刷新 db_hero 对象，确保获取最新的数据库状态
    return db_hero

用 HeroPublic 读取 Hero
我们可以像之前一样读取 Hero 。同样，使用 response_model=list[HeroPublic] 确保正确地验证和序列化数据

@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

用 HeroPublic 读取单个 Hero

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id) # get(要查询的模型类，主键值)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

用 HeroUpdate 更新单个 Hero

我们可以更新单个 hero 。为此，我们会使用 HTTP 的 PATCH 操作

在代码中，我们会得到一个 dict ，其中包含客户端发送的所有数据，只有客户端发送的数据，并排除了任何一个仅仅作为默认值存在的值。为此，我们使用 exclude_unset=True

然后我们会使用 hero_db.sqlmodel_update(hero_data) ，来利用 hero_data 的数据更新 hero_db

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(**hero_data) # 使用 sqlmodel_update 方法更新字段
    session.add(hero_db) # 添加到会话中，标记为已修改
    session.commit()
    session.refresh(hero_db)
    return hero_db