FastAPI

示例代码 服务器向客户端写入Cookie数据 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 @app.get("/async_set_cookie") async def set_cookie_to_client(response: Response): """ 将 cookie 写入到本地文件: test_cookies curl -c test_cookies -X 'GET' \ 'http://127.0.0.1:18081/async_set_cookie' \ -H 'accept: application/json' cat test_cookies # Netscape HTTP Cookie File # https://curl.se/docs/http-cookies.html # This file was generated by libcurl! Edit at your own risk. 127.0.0.1 FALSE / FALSE 0 test_cookie_user_id test_cookie_user_value """ response.set_cookie(key = 'test_cookie_user_id', value = 'test_cookie_user_value') return {"set_cookie": datetime.datetime.now().isoformat(sep = ' ')} 向客户端设置一个名为 test_cookie_user_id 的 cookie，并返回当前的时间戳。 ...

常见类型示例 示例代码 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 from __future__ import annotations import random import string from datetime import date, datetime, time, timedelta from typing import Annotated from uuid import UUID import uvicorn from fastapi import Body, FastAPI from pydantic import BaseModel, EmailStr, HttpUrl app = FastAPI() class MyDataTypes(BaseModel): name: str = ''.join(random.choices(['Alice', 'Bob', 'Charil'], k = 1)) age: int = random.randint(18, 80) price: float = random.randint(100, 1000) / random.randint(1, 100) / 1.0 is_active: bool = False url: HttpUrl = HttpUrl('https:' + '//' + ''.join(random.choices(string.ascii_lowercase + string.digits, k = 16)) + '.com') email: EmailStr = 'test@example.com' unique_id: UUID = UUID('00000000-0000-0000-0000-000000000000') start_datetime: datetime = datetime.now() date_value: date = date.today() process_after: timedelta = timedelta(days = random.randint(1, 10), hours = random.randint(1, 24), minutes = random.randint(1, 59), seconds = random.randint(1, 59)) repeat_at: time = time(hour = random.randint(1, 23), minute = random.randint(1, 59), second = random.randint(1, 59)) @app.post("/async") async def async_root(data: Annotated[MyDataTypes, Body(description = '常用的类型', examples = [ MyDataTypes().model_dump() ])]): print('unique_id:', data.unique_id) print('start_datetime:', data.start_datetime) print('date_value:', data.date_value) print('process_after:', data.process_after) print('repeat_at:', data.repeat_at) return {"message": "Hello World", "data": data} if __name__ == '__main__': uvicorn.run(app, host = '127.0.0.1', port = 18081) 说明 在 FastAPI 中，使用 Pydantic 模型来定义和验证数据是一种常见的做法。 ...

使用多个Model作为Body参数 Model参数定义： 1 2 3 4 5 6 7 8 9 10 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 接受多个 Model 作为 Body 参数的接口定义： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 @app.put("/items/{item_id}") async def update_item(item_id: int, item: Item, user: User): """ curl -X 'PUT' \ 'http://127.0.0.1:18081/items/12345' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{ "item": { "name": "nn", "description": "dd", "price": 1, "tax": 2 }, "user": { "username": "uu", "full_name": "ff" } }' """ results = {"item_id": item_id, "item": item, "user": user} return results ...

Pydantic Model类型 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 class Genders(str, Enum): Male = 'male' Female = 'female' class ExtendItem(BaseModel): a: str = '' b: int = 0 c: float = 0.1 d: bool = False class Config: """ 在 Pydantic 模型中，Config 类用于配置模型的行为。你可以在这个类中设置多种属性来调整模型的解析、验证和序列化行为。以下是一些常用的 Config 类字段： title: 用于为模型提供一个标题，通常用于生成的文档或模式中。 extra: 控制如何处理未在模型中声明的额外字段。可选值包括 ignore、allow、forbid。 arbitrary_types_allowed: 允许模型接受任意类型的字段，而不仅限于标准的 Pydantic 类型。 json_encoders: 为特定类型提供自定义的 JSON 编码器。 alias_generator: 生成别名的函数，用于字段名称，通常用于生成符合特定API规范的别名。 allow_population_by_field_name: 允许通过字段名称而不是别名来填充模型数据。 min_anystr_length: 字符串和字节类型字段的最小长度。 max_anystr_length: 字符串和字节类型字段的最大长度。 min_number_size: 数字类型字段的最小值。 max_number_size: 数字类型字段的最大值。 validate_assignment: 设置为 True 时，将在字段值被赋予新值后触发验证。 error_msg_templates: 自定义错误消息模板。 orm_mode: 允许模型与 ORM 模型兼容，通过允许使用非字典对象进行模型初始化。 use_enum_values: 当设置为 True 时，枚举字段将输出枚举成员的值而不是枚举成员本身。 anystr_strip_whitespace: 自动去除任何字符串或字节字段周围的空白。 schema_extra: 允许为 Pydantic 模型的 JSON Schema 添加额外的信息。 json_loads 和 json_dumps: 自定义 JSON 加载和转储函数。 keep_untouched: 保持某些类型在模型处理中不被修改，通常用于装饰器等高级用途。 """ extra = "allow" # 允许额外字段 class Item1(BaseModel): name: str = 'Bob' description: str price: float tax: int tags: list[str] = [] gender: Genders flags: Literal['test1', 'test2'] class Config: title = 'Item1' extra = "allow" # 允许额外字段 Model的使用 1 2 3 4 5 6 7 8 9 10 11 12 13 test: Item1 = Item1(name = 'Boc', description = 'test_desc', price = 1.23, tax = 123, tags = ['tag_1', 'tag_2'], gender = Genders.Male, flags = 'test1', extend = ExtendItem(a = 'a', b = 1, c = 0.1, d = True)) # 将 json string 转化为 model 对象 test = Item1.model_validate_json(test.model_dump_json(indent = 4)) # 将model对象打印成 json string print(test.model_dump_json(indent = 4)) GET请求的查询参数使用 model 类 1 2 3 4 5 6 7 8 9 10 11 12 13 @app.get("/get_model") async def get_model(items: Annotated[Item1, Query(description = 'GET 请求参数模型定义')]): """ GET请求不支持参数的嵌套 curl -X 'GET' \ 'http://127.0.0.1:18081/get_model?name=Bob&description=desc_test&price=1.23&tax=123&tags=tag_1&tags=tag_2&gender=male&flags=test1' \ -H 'accept: application/json' """ print(items) return { "method": "GET", "message": items } ...

设置查询参数是否为必填 使用Annotated装饰查询参数 不设置default值 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 @app.get("/validation") async def async_root(str_param: Annotated[str | None, Query(min_length = 3, max_length = 20, pattern = '^[a-zA-Z0-9]+$', description = 'This is a string parameter')]): """ 如果不设置 str_param = None, 则 str_param 为必填 """ ret = {"str_list": str_param} if str_param is not None: h = hashlib.sha3_512(str_param.encode('utf-8')).hexdigest() ret.update({"sha3_512": h}) else: ret.update({"uuid": uuid.uuid4().hex}) return ret 不设置默认值 生成的文档中参数为必填 ...