1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

from typing import Union,Annotated,Literal
from fastapi import FastAPI,Query,Path
from pydantic import BaseModel,Field

# Pydantic数据模型会自动进行自动数据校验，并且自动转换数据类型，比如前端传"3.14"，会自动转成浮点数

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

# model_config 是 Pydantic V2 中用于配置模型行为的属性（替代了 Pydantic V1 中的 Config 类）。这里的 {"extra": "forbid"} 作用是禁止接收FilterParams中未定义的字段
class FilterParams(BaseModel):
    model_config = {"extra": "forbid"}
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []

# 请求体
# 与声明查询参数一样，包含默认值的模型属性是可选的，否则就是必选的。默认值为 None 的模型属性也是可选的。
@app.post("/items/")
async def create_item(item: Item):
    return item

# 路径和下面的接口相同，一定要写在上面
@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

# 路径参数
@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

# 查询参数 ?k=v&k=v
# 参数有默认值，因此可选
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

# Union表明或，数据类型要么是str要么为None，即使 q 是可选的，但只要提供了该参数，则该参数值不能超过50个字符的长度，同时也支持正则表达式
@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50,pattern="^fixedquery$")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# Path的title是参数说明，会展示在api文档中，int是类型，Path是元数据校验/约束

@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get", ge=10.5)],
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import time

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
  # 路由到接口函数前执行
    start_time = time.perf_counter()
    response = await call_next(request)
  # 路由到接口函数后执行
    process_time = time.perf_counter() - start_time
  # 计算接口执行所需时间
    response.headers["X-Process-Time"] = str(process_time)
    return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

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"}