fastapi

https://fastapi.tiangolo.com/zh/learn/

0 快速使用

# 异步框架
from fastapi import FastAPI
from pydantic import BaseModel
app=FastAPI()


class Item(BaseModel):
    name: str
    age:int

@app.get('/')
async def index():
    return {'code':100,'msg':'成功'}


@app.post('/login')
async def login(item:Item):
    print(item.name)
    print(item.age)
    return item

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='第一个fastapi:app',port=8001)

1 路径参数

from fastapi import FastAPI

app = FastAPI()


# 1 路径参数---基本使用
@app.get('/home/{id}')
async def home(id: str):
    return {'code': 100, 'id': id}


# 2 路径覆盖-->路由从上往下执行
# 这个永远不会走到
@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}


# 3 枚举类型
from enum import Enum
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    # 1 比较枚举元素

    # 2 获取枚举类型的值
    # 使用 model_name.value 或 your_enum_member.value 获取实际的值
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}
    # 3 返回枚举类型，直接返回对象即可
    print(type(model_name))
    return {"model_name": model_name, "message": "Have some residuals"}


#  4 路径转换器 ############
# 参数名为 file_path，结尾部分的 :path 说明该参数应匹配路径
# /home/johndoe/myfile.txt 的路径参数要以斜杠（/）开头
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

if __name__ == '__main__':
    import uvicorn

    uvicorn.run('路径参数:app', reload=True, port=8080)

2 查询参数

# 除了路径参数以外的，都是查询参数
@app.get("/items2/{item_id}")
async def read_item2(item_id: str, q: str|None = None, short: bool = False):
# 如果有多个路径参数，会依次排，后续都是查询参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False
):

from fastapi import FastAPI
from typing import Union
app = FastAPI()
# 1 查询参数---》如果在路径中没加 {}--->视图函数中得所有参数，都是请求参数
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]


# 2 可选参数  q: str | None = None  ：路径参数+查询参数，查询参数可以不传--->路径和查询参数跟视图函数变量位置无关，但是一般咱们会把路径参数写前面
@app.get("/items1/{item_id}")
# async def read_item1(item_id: str, q: str | None = None):
async def read_item1(q: str | None, item_id: str):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

####3 查询参数类型转换 ######
@app.get("/items2/{item_id}")
async def read_item2(item_id: str, q: str|None = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item


######## 4 多个路径和查询参数##########
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item


#### 5 必选查询参数 ####
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    #1 没设置默认值，参数必须填
    item = {"item_id": item_id, "needy": needy}
    return item

@app.get("/items3/{item_id}")
async def read_user_item3(
    item_id: str, needy: str, skip: int = 0, limit: Union[int, None] = None
):
    # 2 skip 和limit可以不填
    '''
    needy，必选的 str 类型参数
    skip，默认值为 0 的 int 类型参数,可以不填
    limit，可选的 int 类型参数
    '''
    item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
    return item


if __name__ == '__main__':
    import uvicorn

    uvicorn.run('查询参数:app', reload=True, port=8081)

3 请求体

from fastapi import FastAPI
from typing import Union
from pydantic import BaseModel
app = FastAPI()


# 定义一个请求体结构：类
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
# 1 基本使用
@app.post("/items/")
async def create_item(item: Item): #如果参数 不是路径参数，而是使用BaseModel定义的类，都是请求体参数，其他的才是查询参数
    print(type(item))
    print(item.name)
    return item  # 直接返回对象，之间做序列化---》继承BaseModel


# 2 使用模型 在路径操作函数内部直接访问模型对象的属性
@app.post("/items1/")
async def create_item(item: Item):
    item_dict = item.dict() # 对象转字典  item.model_dump
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict  # 返回字典

# 3 请求体 + 路径参数
@app.put("/items2/{item_id}")
async def update_item2(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}


# 4 请求体 + 路径参数 + 查询参数
'''
函数参数按如下规则进行识别：
路径中声明了相同参数的参数，是路径参数
类型是（int、float、str、bool 等）单类型的参数，是查询参数
类型是 Pydantic 模型的参数，是请求体
'''
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: Union[str, None] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result
if __name__ == '__main__':
    import uvicorn

    uvicorn.run('请求体:app', reload=True, port=8081)

4 查询参数和字符串校验

from fastapi import FastAPI, Query,Path,Body
from typing import Union
from pydantic import BaseModel
app = FastAPI()

# 1 查询参数 q 的类型为 str，默认值为 None，因此它是可选的
@app.get("/items/")
async def read_items(q: Union[str, None] = None):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# 2 约束条件：即使 q 是可选的，但只要提供了该参数，则该参数值不能超过50个字符的长度
@app.get("/items1/")
async def read_items1(q: Union[str, None] = Query(default=None, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 3 约束条件：即使 q 是可选的，但只要提供了该参数，则该参数值不能超过50个字符的长度，最短3
@app.get("/items2/")
async def read_items2(q: Union[str, None] = Query(default=None, min_length=3,max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 4 添加正则表达式
@app.get("/items3/")
async def read_items3(
    q: Union[str, None] = Query(
        default=None, min_length=3, max_length=50, pattern="^fixedquery$"
    ),
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 5 默认值
@app.get("/items4/")
async def read_items4(q: str = Query(default="fixedquery", min_length=3)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 6 声明为必需参数
@app.get("/items5/")
async def read_items5(q: str = Query(min_length=3)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# 7 使用省略号(...)声明必需参数
'''
如果你之前没见过 ... 这种用法：它是一个特殊的单独值，它是 Python 的一部分并且被称为「省略号」。 Pydantic 和 FastAPI 使用它来显式的声明需要一个值
'''
@app.get("/items6/")
async def read_items6(q: str = Query(default=..., min_length=3)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 8 使用None声明必需参数
@app.get("/items7/")
async def read_items7(q: Union[str, None] = Query(default=..., min_length=3)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 9 使用Pydantic中的Required代替省略号(...)
# from pydantic import Required  # v2 版本的pydantic 移除了该参数
# @app.get("/items/")
# async def read_items(q: str = Query(default=Required, min_length=3)):
#     results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
#     if q:
#         results.update({"q": q})
#     return results

# 10 查询参数列表 / 多个值
from typing import List, Union
@app.get("/items8/")
async def read_items8(q: Union[List[str], None] = Query(default=None)):
    query_items = {"q": q}
    return query_items
# http://localhost:8000/items/?q=foo&q=bar
# 要声明类型为 list 的查询参数，如上例所示，你需要显式地使用 Query，否则该参数将被解释为请求体


# 11 具有默认值的查询参数列表 / 多个值
from typing import List, Union
@app.get("/items9/")
async def read_items9(q: List[str] = Query(default=["foo", "bar"])):
    query_items = {"q": q}
    return query_items

# 12 也可以直接使用 list 代替 List [str]
@app.get("/items10/")
async def read_items10(q: list = Query(default=[])):
    query_items = {"q": q}
    return query_items

# 13 声明更多元数据  title   description ---  OpenAPI 的支持程度可能不同
@app.get("/items11/")
async def read_items11(
    q: Union[str, None] = Query(default=None, title="Query string", description="这是描述新消息",min_length=3),
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 14 别名参数
#http://127.0.0.1:8000/items/?item-query=foobaritems
# item-query 不是一个有效的 Python 变量名称
#最接近的有效名称是 item_query。
#但是你仍然要求它在 URL 中必须是 item-query...
#这时你可以用 alias 参数声明一个别名，该别名将用于在 URL 中查找查询参数值
@app.get("/items12/")
async def read_items12(q: Union[str, None] = Query(default=None, alias="item-query")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 15 弃用参数
# 现在假设你不再喜欢此参数。
# 你不得不将其保留一段时间，因为有些客户端正在使用它，但你希望文档清楚地将其展示为已弃用。
# 那么将参数 deprecated=True 传入 Query弃用参数
@app.get("/items13/")
async def read_items13(
    q: Union[str, None] = Query(
        default=None,
        alias="item-query",
        title="Query string",
        description="Query string for the items to search in the database that have a good match",
        min_length=3,
        max_length=50,
        pattern="^fixedquery$",
        deprecated=True,
    ),
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

if __name__ == '__main__':
    import uvicorn

    uvicorn.run('查询参数校验:app', reload=True, port=8081

5 路径参数和数值校验

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

app = FastAPI()


#1 Path 路径函数，声明元数据，可以声明与 Query 相同的所有参数
	# 路径参数总是必需的，因为它必须是路径的一部分。
	# 即使你使用 None 声明路径参数或设置一个其他默认值也不会有任何影响，它依然会是必需参数
	# python 3.10版本，使用Annotated


@app.get("/items/{item_id}")
async def read_items(
        # item_id:int= Path(title="The ID of the item to get") Annotated 第一个参数是类型，第二个参数是等于谁
        item_id: Annotated[int, Path(title="The ID of the item to get")],
        q: Annotated[str | None, Query(alias="item-query")] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


# python 3.8版本，不使用Annotated
@app.get("/items/{item_id}")
async def read_items(
        item_id: int = Path(title="The ID of the item to get"),
        q: Union[str, None] = Query(default=None, alias="item-query"),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


# 2 按需对参数排序
'''
假设你想要声明一个必需的 str 类型查询参数 q。
而且你不需要为该参数声明任何其他内容，所以实际上你并不需要使用 Query。
但是你仍然需要使用 Path 来声明路径参数 item_id。
如果你将带有「默认值」的参数放在没有「默认值」的参数之前，Python 将会报错。
但是你可以对其重新排序，并将不带默认值的值（查询参数 q）放到最前面。
对 FastAPI 来说这无关紧要。它将通过参数的名称、类型和默认值声明（Query、Path 等）来检测参数，而不在乎参数的顺序
'''


@app.get("/items/{item_id}")
async def read_items(q: str, item_id: int = Path(title="The ID of the item to get")):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


# 3 按需对参数排序的技巧
'''
如果你想不使用 Query 声明没有默认值的查询参数 q，同时使用 Path 声明路径参数 item_id，并使它们的顺序与上面不同，Python 对此有一些特殊的语法。
传递 * 作为函数的第一个参数。
Python 不会对该 * 做任何事情，但是它将知道之后的所有参数都应作为关键字参数（键值对），也被称为 kwargs，来调用。即使它们没有默认值
'''


@app.get("/items/{item_id}")
async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


# 4 数值校验(int,float)：大于等于，大于，小于等于
'''
使用 Query 和 Path（以及你将在后面看到的其他类）可以声明字符串约束，但也可以声明数值约束。
像下面这样，添加 ge=1 后，item_id 将必须是一个大于（greater than）或等于（equal）1 的整数
gt：大于（greater than）
ge：大于等于（greater than or equal）
lt：小于（less than）
le：小于等于（less than or equal）
数值校验同样适用于 float 值
'''


@app.get("/items/{item_id}")
async def read_items(
        *, item_id: int = Path(title="The ID of the item to get", ge=1), q: str
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

if __name__ == '__main__':
    import uvicorn

    uvicorn.run('路径参数数据校验:app', reload=True, port=8081)

6 请求体多个参数

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

app = FastAPI()


# 1 混合使用 Path、Query 和请求体参数
# 通过将默认值设置为 None 来将请求体参数声明为可选参数
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.put("/items/{item_id}")
async def update_item(
        item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
        q: str | None = None,
        item: Item | None = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results


# 2 多个请求体参数
'''
{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  }
}

'''


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


class User(BaseModel):
    username: str
    full_name: str | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results


# 3 请求体中的单一值 Body 使用
@app.put("/items3/{item_id}")
async def update_item3(
        item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results


# 4 多个请求体参数和查询参数 Body 同样具有与 Query、Path 以及其他后面将看到的类完全相同的额外校验和元数据参数。
@app.put("/items/{item_id}")
async def update_item(
        *,
        item_id: int,
        item: Item,
        user: User,
        importance: Annotated[int, Body(gt=0)],
        q: str | None = None,
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    if q:
        results.update({"q": q})
    return results


# 5 嵌入单个请求体参数
'''
{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  }
}
而不是
{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

'''


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results


if __name__ == '__main__':
    import uvicorn

    uvicorn.run('请求体多个参数:app', reload=True, port=8081)

7 请求体字段校验

from fastapi import FastAPI, Path, Query,Body
from typing import Annotated
from typing import Union
from pydantic import BaseModel ,Field
app = FastAPI()

# 1 使用 pydantic的 Field，声明模型属性
class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None, title="The description of the item", max_length=300
    )
    price: float = Field(gt=0, description="The price must be greater than zero")
    tax: float | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

8 请求体嵌套

from fastapi import FastAPI, Path, Query, Body
from typing import Annotated
from typing import Union,List
from pydantic import BaseModel, Field, HttpUrl
from typing import List, Union

app = FastAPI()

# 1 List 字段
# tags 成为一个由元素组成的列表。不过它没有声明每个元素的类型
'''
{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": []
}
'''


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list = []


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results


# 2 具有子类型的 List 字段
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
    tags: List[str] = []


# 3 Set 类型 标签不应该重复，它们很大可能会是唯一的字符串
'''
即使你收到带有重复数据的请求，这些数据也会被转换为一组唯一项。
而且，每当你输出该数据时，即使源数据有重复，它们也将作为一组唯一项输出
'''

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()

# 4 嵌套模型
'''
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [],
  "image": {
    "url": "string",
    "name": "string"
  }
  "images": [{
    "url": "string",
    "name": "string"
  },{
    "url": "string",
    "name": "string"
  }
  
  ]
}
'''


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()
    image: Image | None = None
    images: List[Image] | None = None


# 5 特殊的类型和校验
# Pydantic 的外部类型 的文档 https://docs.pydantic.dev/latest/concepts/types/#handling-custom-generic-classes
# 在 Image 模型中我们有一个 url 字段，我们可以把它声明为 Pydantic 的 HttpUrl，而不是 str
class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()
    image: Image | None = None


# 6 有一组子模型的属性 
'''
{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": [
        "rock",
        "metal",
        "bar"
    ],
    "images": [
        {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        },
        {
            "url": "http://example.com/dave.jpg",
            "name": "The Baz"
        }
    ]
}
'''


class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()
    images: list[Image] | None = None  # 有一组子模型的属性 


# 6 深度嵌套 请注意 Offer 拥有一组 Item 而反过来 Item 又是一个可选的 Image 列表是如何发生的
class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()
    images: list[Image] | None = None


class Offer(BaseModel):
    name: str
    description: str | None = None
    price: float
    items: list[Item]


@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer


# 7 纯列表请求体
class Image(BaseModel):
    url: HttpUrl
    name: str


@app.post("/images/multiple/")
async def create_multiple_images(images: list[Image]):  # 纯列表请求体
    return images


# 8 任意 dict 构成的请求体
'''
请记住 JSON 仅支持将 str 作为键。
但是 Pydantic 具有自动转换数据的功能。
这意味着，即使你的 API 客户端只能将字符串作为键发送，只要这些字符串内容仅包含整数，Pydantic 就会对其进行转换并校验。
然后你接收的名为 weights 的 dict 实际上将具有 int 类型的键和 float 类型的值
'''

@app.post("/index-weights/")
async def create_index_weights(weights: dict[int, float]):
    return weights

9 模式的额外信息和其他字段

from fastapi import FastAPI, Path, Query,Body
from typing import Annotated
from typing import Union
from pydantic import BaseModel ,Field,HttpUrl
from typing import List, Union
app = FastAPI()

# 1 Pydantic schema_extra
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None

    # 这些额外的信息将按原样添加到输出的JSON模式中,作为api的提示
    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ]
        }
    }


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results
  
# 2 Field 的附加参数
 ## 在 Field, Path, Query, Body 和其他你之后将会看到的工厂函数，你可以为JSON 模式声明额外信息，你也可以通过给工厂函数传递其他的任意参数来给JSON 模式声明额外信息，比如增加 example
class Item(BaseModel):
    name: str = Field(examples=["Foo"])
    description: str | None = Field(default=None, examples=["A very nice Item"])
    price: float = Field(examples=[35.4])
    tax: float | None = Field(default=None, examples=[3.2])


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results
  
# 3 Body 额外参数
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[
        Item,
        Body(
            examples=[
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ],
        ),
    ],
):
    results = {"item_id": item_id, "item": item}
    return results

10 form表单

# pip install python-multipart
import uvicorn
from fastapi import FastAPI,Form,Body

app = FastAPI()

@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

if __name__ == '__main__':
    uvicorn.run("form表单:app", port=8080)

11 文件上传

import uvicorn
from typing import List
from fastapi import FastAPI,Form,File,UploadFile

app = FastAPI()

# 小文件上传
@app.post("/file/")
async def create_file(file: bytes = File()):
    print(file) #字节流数据，小文件适合
    with open('a.ini','wb') as f:
        f.write(file)
    return {"file_size": len(file)}
# 大文件上传
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    print(file) #文件句柄方式，适合大文件
    # with open(file.filename,'wb') as f:
    #     data = await file.read()
    #     f.write(data)

    with open(file.filename, "wb") as buffer:
        while True:
            data = await file.read(1024)
            if not data:
                break
            buffer.write(data)
    return {"filename": file.filename}

# 多文件上传
@app.post("/files/")
async def create_files(files: List[bytes] = File()):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile]):
    return {"filenames": [file.filename for file in files]}
if __name__ == '__main__':
    uvicorn.run("s:app", host="127.0.0.1", port=5000,reload=True)

12 Request对象

import uvicorn
from typing import List
from fastapi import FastAPI,Request

app = FastAPI()

# 小文件上传
@app.post("/demo/{item_id}")
async def demo(item_id:int,request:Request,q:str=None):
    print(item_id)
    print(q)
    print(request)
    # 1 cookie
    print(request.cookies)
    # 2 header:user-agent
    print(request.headers)  # user-agent在里面
    print(request.headers.get('user-agent'))
    # 3 url
    print(request.url)
    # 4 请求体数据
    from starlette.requests import Request
    # data = await request.body()
    # data = await request.body()
    data = await request.json()
    print(data)

    # 5 客户端ip地址和端口
    print(request.client.host)
    print(request.client.port)

    # 6 请求方式
    print(request.method)



    return 'ok'

if __name__ == '__main__':
    uvicorn.run("request对象:app", host="127.0.0.1", port=5000,reload=True)

13 前端传的cookie和Header

# 1 基本使用
##3.9+ Annotated
from typing import Annotated, Union
from fastapi import Cookie, FastAPI,Request,Header
import uvicorn
app = FastAPI()
'''
python 3.8  --没有Annotated
async def read_items(ads_id: Union[str, None]= Cookie(default=None)):
python 3.9 --有Annotated
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
python 3.10 --有Annotated 有 |
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
'''
@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
    return {"ads_id": ads_id}

'''
Cookie 、Path 、Query 是兄弟类，都继承自共用的 Param 类。

注意，从 fastapi 导入的 Query、Path、Cookie 等对象，实际上是返回特殊类的函数。

必须使用 Cookie 声明 cookie 参数，否则该参数会被解释为查询参数
'''

# postman 访问，在请求头的cookie参数中写入 ads_id=9001


# 2 同样可以在request对象中取出来
@app.get("/items1/")
async def read_items1(request:Request,ads_id: Annotated[Union[str, None], Cookie()] = None):
    c=request.cookies
    print(c)
    return {"ads_id": ads_id}

if __name__ == '__main__':
    uvicorn.run("获取cookie:app", host="127.0.0.1", port=5000,reload=True)

14 响应模型参数

就是序列化类和反序列化类不使用同一个

read_only和write_only的控制

import uvicorn
from typing import List,Union,Any
from pydantic import BaseModel,EmailStr
from fastapi import FastAPI,Form,File,UploadFile,Request

app = FastAPI()


# 1 响应跟传入的相同
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None

# 2 不指定
@app.post("/user/")
async def create_user(user: UserIn) -> UserIn:
    return user


# 3 response_model 使用
@app.post("/user2/",response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user

# 4 response_model_exclude_unset 使用  使用 response_model_exclude_unset 来仅返回显式设定的值
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []
items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]
# 5 还可以使用：
# '''
# response_model_exclude_defaults=True
# response_model_exclude_none=True
# response_model_include
# response_model_exclude
# '''
@app.get("/items1/{item_id}", response_model=Item, response_model_exclude=['name'])
async def read_item1(item_id: str):
    return items[item_id]

if __name__ == '__main__':
    uvicorn.run("响应模型参数:app", host="127.0.0.1", port=5000,reload=True)

15 配置文件

15.1 settings.py方案

完全自己定义，导入使用

15.2 dotenv 方案

# 1 安装python-dotenv
# python-dotenv 是一个Python库，用于从文本文件中加载环境变量
pip3 install python-dotenv
# 2 安装pydantic_settings
pip3 install pydantic_settings
# 3 编写 .env
# -------- 服务配置信息 --------
APP_PORT=8080
APP_HOST=0.0.0.0
APP_VERSION=v1.0.0
APP_DEBUG=true

# -------- 数据库配置 --------
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123
DB_DATABASE=demo001

# -------- redis配置 --------
REDIS_DSN=redis://127.0.0.1:6379/0
        
# 4 编写配置模型 config.py

from pydantic import RedisDsn
from pydantic_settings import BaseSettings
from dotenv import load_dotenv


class AppConfigSettings(BaseSettings):
    """应用配置"""
    app_port: int = 8080
    app_env: str = "dev"
    app_debug: bool = False
    app_host: str
    """数据库配置"""
    db_host: str
    db_port: int
    db_user: str
    db_password: str
    db_database: str
    """redis配置"""
    redis_dsn: RedisDsn = None


# 执行---》把 .env 中配置--》映射到 配置类中AppConfigSettings
load_dotenv()
# 实例化配置模型
appSettings = AppConfigSettings()

        
        
# 5 main.py 中使用
import uvicorn
from fastapi import FastAPI
app = FastAPI()
from config import appSettings  # 天然单例
@app.get("/")
async def root():
    print(appSettings.db_database)
    return {"message": "Hello World"}

if __name__ == '__main__':
    uvicorn.run("配置文件:app",reload=True,port=appSettings.app_port,host=appSettings.app_host)

16 路由分发

# fastapi APIRouter--》路由分发--》目录划分
	-1 实例化得到 APIRouter对象
    -2 在app中注册
    -3 以后写路径都用，APIRouter对象注册路径即可

1 小型项目

"""
├── fastapi_x
	├── src/				# 项目运行时/开发时日志目录 - 包
		├── home.py   	
		├── models.py   	# 表模型
		└── users.py  		
	├── .env 		        # 其他配置信息
    ├── main.py 		    # 脚本文件
    └── settings.py    		# 配置文件
"""

main.py

from fastapi import FastAPI
from src.home import home
from src.users import user
app = FastAPI()

app.include_router(home,prefix='/api/v1/home',tags=['主页相关接口'])
app.include_router(user,prefix='/api/v1/user',tags=['用户相关接口'])

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main:app',reload=True)

settings.py

from pydantic import RedisDsn
from pydantic_settings import BaseSettings
from dotenv import load_dotenv
class AppConfigSettings(BaseSettings):
    """应用配置"""
    app_port: int = 8080
    app_env: str = "dev"
    app_debug: bool = False
    """数据库配置"""
    db_host: str
    db_port: int
    db_user: str
    db_password: str
    db_database: str
    """redis配置"""
    redis_dsn: RedisDsn = None

# 加载 .env 文件
load_dotenv()
# 实例化配置模型
settings = AppConfigSettings()

.env

# -------- 服务配置信息 --------
APP_PORT=8080
APP_HOST=0.0.0.0
APP_VERSION=v1.0.0
APP_DEBUG=true

# -------- 数据库配置 --------
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123
DB_DATABASE=lqz_mysql

# -------- redis配置 --------
REDIS_DSN=redis://127.0.0.1:6379/0

home.py

from fastapi import APIRouter
from settings import settings

home = APIRouter()


@home.get("/")
async def root():
    print(settings.db_database)
    return {"message":settings.db_database}


@home.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}

users.py

from fastapi import APIRouter

user = APIRouter()


@user.get("/users")
async def root():
    return {"message": "返回所有用户"}

2 大型项目

"""
├── fastapi_x
	├── apps/				# 项目运行时/开发时日志目录 - 包
		├── home/  	
		├── users/
		...
	├── .env 		        # 其他配置信息
    ├── main.py 		    # 脚本文件
    └── settings.py    		# 配置文件
"""

17 中间件

import asyncio
import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()


# @app.middleware("http")
# async def add_process_time_header(request: Request, call_next):
#     # 1 统计时间---》process_request
#     start_time = time.time()
#     # 2 继续执行下面的中间件---》继续执行视图函数
#     response = await call_next(request)
#     # 3 视图函数执行完，取了时间--》process_response
#     process_time = time.time() - start_time
#     response.headers["X-Process-Time"] = str(process_time)
#     return response

# 请求头中必须带token，如果不带，直接返回
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    token = request.headers.get('token', None)
    if token:
        response = await call_next(request)
        
        return response
    else:
        return JSONResponse({'code': 101, 'msg': '没有携带token'})


@app.get('/index')
async def index():
    await asyncio.sleep(1)  # 模拟io
    return 'index'


if __name__ == '__main__':
    import uvicorn

    uvicorn.run('中间件:app', reload=True, port=8888)

18 cors

import uvicorn
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='*',  #  * 表示所有
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


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

if __name__ == '__main__':
    uvicorn.run("cors:app", host="127.0.0.1", port=5000,reload=True)

19 模板和静态文件

import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles

app = FastAPI()
# 静态文件
app.mount("/static", StaticFiles(directory="static"))
# 模板
templates = Jinja2Templates(directory="templates")


@app.get("/index")
def index(request: Request):
    context = {
        'request': request,
        'name': 'lqz',
        "age": 19,
        "hobby": ["篮球", "足球"],
        "info": {
            "city": "上海",
            "address": "浦东"
        }
    }
    return templates.TemplateResponse('index.html', context)


if __name__ == '__main__':
    uvicorn.run("静态文件和模板:app", host="127.0.0.1", port=5000,reload=True)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<h1>{{name}} ({{age}})</h1>
<ul>
    {% for item in hobby%}
    <li>{{item}}</li>
    {% endfor %}
</ul>
<div>
    <ul>
        {% for k,v in info.items()%}
        <li>{{k}}{{v}}</li>
        {% endfor %}
    </ul>
</div>
<img src="/static/demo.png" style="width: 200px;">
</body>
</html>

20 ORM

# 0 官方文档：sqlalchemy，普遍用tortoise更多一些

# 1 文档
# github:https://github.com/tortoise/tortoise-orm
# 文档：https://tortoise.github.io/getting_started.html

# 2 安装
pip3 install tortoise-orm[asyncmy]

创建表

from tortoise import Model,fields

class Publish(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255)
    addr = fields.CharField(max_length=255)
    def __str__(self):
        return self.name


class Book(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255)
    publish = fields.ForeignKeyField('models.Publish', related_name='book')
    authors = fields.ManyToManyField('models.Author', related_name='books', through='author_book')

    def __str__(self):
        return self.name


class Author(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255)

    def __str__(self):
        return self.name

main.py

from fastapi import FastAPI
from apps.home.views import home
from apps.order.views import order

from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORM
app = FastAPI()
# register_tortoise(app,
#                   db_url="mysql://root:[email protected]:3306/fastapi",
#                   modules={"models": ["apps.home.models","apps.order.models"]},  # models列表路径
#                   )
register_tortoise(app,config=TORTOISE_ORM)
app.include_router(home,prefix='/api/v1/home',tags=['home描述信息'])
app.include_router(order,prefix='/api/v1/order',tags=['order描述信息'])

settings.py

TORTOISE_ORM = {
    'connections': {
        'default': {
            # 'engine': 'tortoise.backends.asyncpg',  PostgreSQL
            'engine': 'tortoise.backends.mysql',  # MySQL or Mariadb
            'credentials': {
                'host': '127.0.0.1',
                'port': '3306',
                'user': 'root',
                'password': '123',
                'database': 'fastapi',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                "echo": True
            }
        },
    },
    # "connections": {"default": "mysql://root:[email protected]:3306/fastapi"},
    'apps': {
        'models': {
            # models:models 找到对应自定义的model.py
            'models': ['apps.home.models','apps.order.models', "aerich.models"],  # aerich.models迁移模型
            'default_connection': 'default',
        }
    },
    'use_tz': False,
    'timezone': 'Asia/Shanghai'
}

aerich迁移

# 0 pip3 install aerich
# 1. 打开命令行，切换到项目根目录 
cd 项目
# 2. 初始化配置项
aerich init -t settings.TORTOISE_ORM

    #初始化完成后会在当前目录生成一个文件pyproject.toml和一个文件夹migrations
    #· pyproject.toml: 保存配置文件路径
    #. migrations：存放.sql迁移文件 
    
# 3. 初始化数据库，一般情况下只用一次
aerich init-db
    
    #· 如果TORTOISE_ORM配置文件中的models改了名，执行这条命令时需要增加--app参数，来指定你修改的名字。
    #· 在migrations的指定app目录下生成sql文件（如果model不为空时），并在数据库中生成表。

    
# 4. 更新模型并进行迁移---更新使用
aerich migrate --name any_name
    
    #· 每次修改model后执行此命令，将在migrations文件夹下生成.sql迁移文件；
    #· --name参数为你的迁移文件添加备注，默认为update；
    #· 迁移文件名的格式为{version_num}{datetime}{any_name|update}.sql；
    #· 如果aerich识别到您正在重命名列，它会要求重命名{old_column}为{new_column} [True]，您可以选择True
    #  重命名列而不删除列，或者选择False删除列然后创建，如果使用Mysql，只有8.0+支持重命名。
       
# 5. 更新最新修改到数据库
aerich upgrade    或aerich upgrade [xxx.sql]


# 6. 后续每次修改models文件内容（新增、删除或修改数据模型， 非models文件重命名），只需执行步骤4、5即可.

# 7. 其它操作
# 降级到指定版本
aerich downgrade -v 指定版本 -d 降级的同时删除迁移文件 --yes 确认删除，不再交互式输入

# 显示当前可以迁移的版本
aerich heads

# 显示迁移历史
aerich history

增删查改

# 查询所有
# books = await Book.all()
# 过滤
# books = await Book.filter(name='红楼')
# 模糊查询
# books = await Book.filter(name__contains='红')
# books = await Book.filter(id__gt=1).all()
# 查询值
# books = await Book.filter(name='红楼').values('name', 'id')

# 获取单个对象
# book = await Book.get(name='红楼')

# 多对多查询
books = await Book.filter(book__publish__name="北京出版社").values('name')

# 一对多查询
# books = await Book.filter(book__authors__name='张三').values('name')





# 多对多查询
books = await Book.filter(authors__name="张三").values('name')

# 一对多查询
books = await Book.filter(publish__name="北京出版社").values('name')

 # 方式一
book = Book(name='西游记', publish_id=1)
await book.save()

# 方式二
book = await Book.create(name='三国演义', publish_id=2)

# 多对多关系
book = await Book.get(name='三国演义')
publish = await Publish.get(name='北京出版社')
await book.publish.add(publish)

 # 方式一
book = await Book.get(name='红楼')
book.name = '红楼1'
await book.save()

# 方式二
await Book.filter(name='红楼').update(name='红楼梦')

 # 方式一
book = await Book.get(name='红楼')
await book.delete()

# 方式二
await Book.filter(name='红楼').delete()

https://blog.csdn.net/weixin_54217348/article/details/136772215
https://www.cnblogs.com/pearlcity/p/17829736.html
https://github.com/prostomarkeloff/fastapi-tortoise

21 websocket

from fastapi import FastAPI, WebSocket,Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
import uvicorn
app = FastAPI()
# 静态文件
app.mount("/static", StaticFiles(directory="static"))
# 模板
templates = Jinja2Templates(directory="templates")
html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8001/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


@app.get("/")
async def get():
    return HTMLResponse(html)

@app.get('/index')
def index(request:Request):
    context={
        'request':request
    }
    return templates.TemplateResponse('websocket.html',context)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    count=0
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

if __name__ == '__main__':
    uvicorn.run("websocket:app", host="127.0.0.1", port=8001,reload=True)