Corn 的自留地

好久没有学习 python 的新技术了。今天刷到了 python 的讲解 __init__.py 用处的视频和讲解使用 uv 项目管理工具的视频，觉得 python 也可以做到工程化的开发。

正好目前准备学习 WebSocket ，我又无意中知道了 FastAPI ，这里做个笔记记录一下用法，免得后面忘了。

现在我的理解就是开发常规 app 锻炼设计能力还是用 python 比较方便，而一些高性能的算法可以使用 Rust 为 python 写库，提高性能。（而且我意识到 Rust 真的很适合写库，因为库的需求就是高性能，稳定，安全，而 Rust 都满足这些东西。而常规的 app 这种需求经常更改的东西使用 Rust 写会非常痛苦。还有 Rust 写底层的一些程序和开发一些高性能的工具也是非常适合的）

FastApi 最吸引我的地方就是它可以自动生成 API 文档，实在太方便了。

而且原生支持 Restful API，不需要额外的学习成本。

python# 导入包
import requests
import json
import os

参数

pythonfrom sys import prefix
from fastapi import APIRouter
router = APIRouter(
    prefix="/para",
    tags=["参数"]
)

pythonbase_url = "http://127.0.0.1:8000/para"

常规的路径参数

python@router.get("/users/{user_id}")
async def get_user(user_id: int): # 也可以用Annotated[int, Path()]，左边是参数类型，右边表示是什么类型的参数
    # get user from database or other source
    user = {"name": "John Doe", "age": 30}   # example user data  
    return user

pythonrequests.get(base_url + '/users/1').json()

plaintext    {'name': 'John Doe', 'age': 30}

python# 如果传入的不是符合参数类型的参数
requests.get(base_url + '/users/1.5').json()

plaintext    {'detail': [{'type': 'int_parsing',
       'loc': ['path', 'user_id'],
       'msg': 'Input should be a valid integer, unable to parse string as an integer',
       'input': '1.5'}]}

路径的路径参数方式

python@router.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

pythonrequests.get(base_url + '/files/path/to/file').json()

plaintext    {'file_path': 'path/to/file'}

查询参数

pythonfake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
@router.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

pythonrequests.get(base_url + '/items').json()

plaintext    [{'item_name': 'Foo'}, {'item_name': 'Bar'}, {'item_name': 'Baz'}]

pythonrequests.get(base_url + '/items?skip=1&limit=2').json()

plaintext    [{'item_name': 'Bar'}, {'item_name': 'Baz'}]

pythonfrom typing import Annotated
from fastapi import Header
@router.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    # 会自动把请求头中的user-agent转换为user_agent
    return {"User-Agent": user_agent}

pythonheader = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
}
requests.get(base_url + '/header_para/', headers=header).json()

plaintext    {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'}

还可以用自定义的参数模型，和请求参数 Query 和路径参数 Path 是一样的用法

pythonclass CommonHeaders(BaseModel):
    host: str
    save_data: bool
    if_modified_since: str | None = None
    traceparent: str | None = None
    x_tag: list[str] = []


@router.get("/header_para_with_model/")
async def get_header_para_with_model(headers: Annotated[CommonHeaders, Header()]):
    return headers

表单参数

使用 Form() 就行了。话说现在表单用得很少了吧，一般都是用 JSON 传输数据。

请求体

pythonrouter = APIRouter(
    prefix="/body",
    tags=["请求体"]
)

pythonbase_url = "http://127.0.0.1:8000/body"

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

@router.post("/item/")
async def create_item(item: Item):
    return item

@router.post("/items/")
async def create_item(items: list[Item]):
    return items

pythondata = {
    'name': 'dog',
    'description': 'A cute dog', # optional
    'price': 24.99,
    'tax': 0.08                  # optional
}
data = json.dumps(data)
requests.post(base_url + '/item/1', data).json()

plaintext    {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08}

pythondog = {
    'name': 'dog',
    'description': 'A cute dog', # optional
    'price': 24.99,
    'tax': 0.08                  # optional
}
data = json.dumps([dog for _ in range(10)])
requests.post(base_url + '/items/',data).json()

plaintext    [{'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08},
     {'name': 'dog', 'description': 'A cute dog', 'price': 24.99, 'tax': 0.08}]

额外数据类型

pythonrouter = APIRouter(
    prefix="/data_type",
    tags=["额外数据类型"]
)

pythonbase_url = "http://127.0.0.1:8000/data_type"

pythonfrom datetime import datetime, time, timedelta
from typing import Annotated
from uuid import UUID

from fastapi import Body

@router.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: Annotated[datetime, Body()],
    end_datetime: Annotated[datetime, Body()],
    process_after: Annotated[timedelta, Body()],
    repeat_at: Annotated[time | None, Body()] = None,
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "process_after": process_after,
        "repeat_at": repeat_at,
        "start_process": start_process,
        "duration": duration,
    }

pythonitem_id = "12345678-1234-1234-1234-123412341234"
data = {
    "start_datetime": "2022-01-01T12:00:00Z",
    "end_datetime": "2022-01-01T13:00:00Z",
    "process_after": "00:30:00",
    "repeat_at": "14:00:00",
}

response = requests.put(base_url + f"/items/{item_id}", json=data).json()
# 规范打印json
print(json.dumps(response, indent=4))

plaintext    {
        "item_id": "12345678-1234-1234-1234-123412341234",
        "start_datetime": "2022-01-01T12:00:00+00:00",
        "end_datetime": "2022-01-01T13:00:00+00:00",
        "process_after": 1800.0,
        "repeat_at": "14:00:00",
        "start_process": "2022-01-01T12:30:00+00:00",
        "duration": 1800.0
    }

请求文件

pythonrouter = APIRouter(
    prefix="/file",
    tags=["文件"]
)

pythonbase_url = "http://localhost:8000/file"

单文件上传

pythonfrom fastapi import FastAPI, File, UploadFile

app = FastAPI()

# 占用内存的写法
@app.post("/files/")
async def create_file(file: bytes = File()):
    return {"file_size": len(file)}


# 如果文件过大会写到磁盘里面，无脑选这个
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    file_py = file.file # 得到python中的文件对象
    return {"filename": file.filename}

python# 创建测试文件
with open("test.txt", "w") as f:
    f.write("Hello, world!")

files = {"file": open("test.txt", "rb")}

print(requests.post(base_url + "/uploadfile/", files=files).json())
# 发送完毕后，删除测试文件
os.remove("test.txt")

{'filename': 'test.txt'}

多文件上传

python@router.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}

pythonfiles = []
for file in os.listdir():
    # 如果不是文件夹
    if not os.path.isdir(file):
        # 打开文件
        files.append(
            ("files", open(file, 'rb'))
        )

requests.post(base_url + '/uploadfiles', files=files).json()

plaintext    [('files', <_io.BufferedReader name='uv.lock'>), ('files', <_io.BufferedReader name='makefile'>), ('files', <_io.BufferedReader name='pyproject.toml'>), ('files', <_io.BufferedReader name='FastApi 学习笔记.ipynb'>), ('files', <_io.BufferedReader name='README.md'>), ('files', <_io.BufferedReader name='.gitignore'>), ('files', <_io.BufferedReader name='.python-version'>), ('files', <_io.BufferedReader name='test.txt'>)]    {'filenames': ['uv.lock',
      'makefile',
      'pyproject.toml',
      'FastApi 学习笔记.ipynb',
      'README.md',
      '.gitignore',
      '.python-version',
      'test.txt']}

可在一个路径操作中声明多个 File 与 Form 参数，但不能同时声明要接收 JSON 的 Body 字段。因为此时请求体的编码为 multipart/form-data，不是 application/json。

这不是 FastAPI 的问题，而是 HTTP 协议的规定。

意思就是如果想在传输文件请求中还要传输参数数据，就只能通过Form表单的方式来传输，因为此时请求体的数据格式只能是表单格式 multipart/form-data，而不是json格式。

错误处理

pythonrouter = APIRouter(
    prefix="/error",
    tags=["错误处理"]
)

pythonbase_url = "http://localhost:8000/error"

一般的 Http 错误消息返回

可以通过抛出异常的方式向客户端返回错误

pythonfrom fastapi import HTTPException

items = {"foo": "The Foo Wrestlers"}
@router.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

pythonresponse = requests.get(base_url + f"/items/foo")
print(response.status_code, response.json())

response = requests.get(base_url + f"/items/other")
print(response.status_code, response.json())

plaintext    200 {'item': 'The Foo Wrestlers'}
    404 {'detail': 'Item not found'}

覆盖（自定义）错误处理器

手动抛出的Http异常

pythonfrom fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse({"message": exc.detail}, status_code=exc.status_code)

pythonresponse = requests.get(base_url + f"/items/foo")
print(response.status_code, response.json())

response = requests.get(base_url + f"/items/other")
print(response.status_code, response.json())

plaintext    200 {'item': 'The Foo Wrestlers'}
    404 {'message': 'Item not found'}

FastAPI 自带的参数转换错误异常

pythonfrom fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse({"message": exc.detail}, status_code=exc.status_code)

弄一个指定路径参数为int的api

python@router.get("/items_int/{interger}")
async def read_item_interger(interger: int):
    return {"item": interger}

pythonresponse = requests.get(base_url + f"/items_int/1")
print(response.status_code, response.json())

response = requests.get(base_url + f"/items_int/2.5")
print(response.status_code, response.json())

plaintext    200 {'item': 1}
    400 {'message': "[{'type': 'int_parsing', 'loc': ('path', 'interger'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': '2.5'}]"}

更新数据

pythonrouter = APIRouter(
    prefix="/update",
    tags=["更新数据"]
)

pythonbase_url = "http://localhost:8000/update"

put 更新

python@router.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded: dict = item.model_dump()
    items[item_id]: dict = update_item_encoded
    return update_item_encoded

pythondata = {"name": "baz", "description": "name is baz", "price": 50.2, "tax": 10.5, "tags": []}

response = requests.put(base_url + f"/items/baz", json=data)
print(response.status_code, response.json())

plaintext    200 {'name': 'baz', 'description': 'name is baz', 'price': 50.2, 'tax': 10.5, 'tags': []}

patch 部分更新

python@router.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    # 从数据库中取出需要更改的item数据并转为Item模型
    stored_item_model: Item = Item(**items[item_id])
    # 获取需要更新的字段
    update_data: dict = item.model_dump(exclude_unset=True)
    # 获取更新后的Item模型
    updated_item: Item = stored_item_model.model_copy(update=update_data)
    # 更新数据库
    items[item_id]: dict = item.model_dump()

    return updated_item

pythondata = {"name": "666"}

response = requests.patch(base_url + f"/items/baz", json=data)
print(response.status_code, response.json())

plaintext    200 {'name': '666', 'description': 'name is baz', 'price': 50.2, 'tax': 10.5, 'tags': []}

依赖项

我的理解就是一个自动接收 endpoint 的参数的可调用对象。可以用在一些路由通用的参数获取和处理上。总之就是可以简化那种好几个endpoint都需要获取或者处理的参数处理。

pythonrouter = APIRouter(
    prefix="/depend",
    tags=["依赖项"]
)

pythonbase_url = "http://localhost:8000/depend"

函数作为依赖项

python# 函数作为依赖项
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 使用同样的参数处理
@router.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

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

pythoncommon_para = "?q=python&skip=1&limit=10"
response = requests.get(base_url + f"/items{common_para}")
print(response.status_code, response.json())
response = requests.get(base_url + f"/users{common_para}")
print(response.status_code, response.json())

plaintext    200 {'q': 'python', 'skip': 1, 'limit': 10}
    200 {'q': 'python', 'skip': 1, 'limit': 10}

类作为依赖项

python# 类作为依赖项
class Person(BaseModel):
    # 别被这种类型系统看晕了，实际上意思很简单。
    # 就是，首先，name可以是str类型，也可以是None类型。
    # 然后，对name进行标注，说它是个Query参数，并且长度限制在10到1之间。
    # age同理，不过有一个默认值，即None。意思就是不一定要传age参数，但如果传了，必须在0到120之间。
    name: Annotated[str | None, Query(max_length=10, min_length=1)]
    age: Annotated[int | None, Query(ge=0, le=120)] = None

# 使用同样的类作为输入
@router.get("/persons")
async def get_person(person: Person = Depends(Person)):
    return person

pythoncommon_para = "?name=John"
response = requests.get(base_url + f"/persons{common_para}")
print(response.status_code, response.json())

plaintext    200 {'name': 'John', 'age': None}

弥补了查询参数和header参数无法使用自动转换的缺陷

子依赖项

依赖还可以嵌套，也就是一个依赖项的参数可以使用别的依赖项。

不举例了，需要用的时候自己查文档吧。

路径操作依赖项

这种方式使用的依赖项不会管依赖项的返回值，而只会让依赖项接收掉相应的参数进行相关处理。

利用这种特性，我们可以实现对安全相关的参数的提前处理

python# 路径操作装饰器中使用依赖项
async def verify_token(x_token: Annotated[str, Header(name="X-Token")]):
    if x_token != "token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def verify_key(x_key: Annotated[str, Header(name="X-Key")]):
    if x_key != "key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

@router.get("/things/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_things():
    return {"things": "stuff"}

pythonbad_header = {
    "X-Token": "bad token",
    "X-key": "bad key"
}

good_header = {
    "X-Token": "token",
    "X-key": "key"
}

response = requests.get(base_url + f"/things", headers=bad_header)
print(response.status_code, response.json())

response = requests.get(base_url + f"/things", headers=good_header)
print(response.status_code, response.json())

plaintext    400 {'message': 'X-Token header invalid'}
    200 {'things': 'stuff'}

全局依赖项

pythonapp = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

使用yield的依赖项

感觉不太用得到，需要用的时候再查吧

安全性

fastapi 对于 OAuth2 的集成

pythonrouter = APIRouter(
    prefix="/security",
    tags=["安全"]
)

pythonbase_url = "http://localhost:8000/security"

给端点的加上 OAuth2PasswordBearer 的实例作为依赖项，就会自动提取请求中header中的 Authorization 字段中的token，并验证token的有效性。

pythonoauth2_scheme = OAuth2PasswordBearer(tokenUrl="/security/login")

@router.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

pythonheader = {"Authorization": "Bearer <token>"}

response = requests.get(base_url + f"/items", headers=header)
print(response.status_code, response.json())

no_token_response = requests.get(base_url + f"/items")
print(no_token_response.status_code, no_token_response.json())

plaintext    200 {'token':'<token>'}
    401 {'message': 'Not authenticated'}
    ```

### 不处理token然后直接返回显然是没什么用的，所以下面是如何对token进行处理

```python
# 为了直接从token解析出用户模型，首先需要定义用户模型
class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

# 定义一个自定义依赖项，其中把OAuth2PasswordBearer的实例作为子依赖项
# 我们就能够自定义对token的处理操作
async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 对 token 进行解码返回假的User模型
    user = User(
        username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
    )
    return user


@router.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

pythonheader = {"Authorization": "Bearer <token>"}

response = requests.get(base_url + f"/users/me", headers=header)
print(response.status_code, response.json())

no_token_response = requests.get(base_url + f"/users/me")
print(no_token_response.status_code, no_token_response.json())

plaintext    200 {'username': '<token>fakedecoded', 'email': '[email protected]', 'full_name': 'John Doe', 'disabled': None}
    401 {'message': 'Not authenticated'}
    404 {'message': 'Not Found'}

把客户端发过来的token解析为用户的问题解决了，但是如何让用户的客户端获取token并且自动保存token并且每次请求都带上token呢？

OAuth2 的标准提供了统一的解决方案：

要求客户端使用密码流时，必须以表单格式发送 username 和 password 两个参数（还允许一个自定义字段 scope，通常用于指定权限，用得到的时候再说吧）。

然后，规定服务端对于以上的请求，返回一个 JSON 格式的数据，有两个字段：access_token 和 token_type。

python# 带密码的用户模型，两种模型分开，有助于安全性
class UserInDB(User):
    hashed_password: str

python# 从数据库中获取用户字典的假方法
def get_user_by_username(username: str) -> dict | None:
    if username == "john":
        return {
            "username": "john",
            "email": "[email protected]",
            "full_name": "John Doe",
            "hashed_password": "hashed(John's Password)"
        }
    return None

# 假的密码加密方法
def fake_hash_password(password: str) -> str:
    return "hashed(" + password + ")"

python# 登录 token 功能实现
@router.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 先根据用户名从数据库中获取用户字典
    user_dict: dict | None = get_user_by_username(form_data.username)
    # 如果用户不存在，返回 400 错误
    if not user_dict:
        raise HTTPException(status_code=400, detail="Unknown username")

    # 如果用户存在，验证密码是否正确
    user: UserInDB = UserInDB(**user_dict) # 转换为 UserInDB 模型
    # 对密码进行加密
    hashed_password = fake_hash_password(form_data.password) 
    # 比较加密后的密码和数据库中的密码是否一致
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    # 这里直接把用户名作为 access_token 返回，实际上应该生成一个签名的 token
    return {"access_token": user.username, "token_type": "bearer"}

真正可用的令牌和密钥

下面介绍 jwt 令牌和 Bcrypt哈希密码的使用。

jwt 令牌

使用 python-jose库

python# 安全配置 (通常应放在环境变量中)
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

python# 创建令牌
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    """创建JWT访问令牌"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) # 默认15分钟过期
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# /login 登录接口
"""用户登录获取JWT令牌"""
user = authenticate_user(form_data.username, form_data.password)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
    data={"sub": user.username}, # 一般sub字段存放username
    expires_delta=access_token_expires
)

python# 从令牌中解析出用户
async def get_current_user(token: str = Depends(oauth2_scheme)):
    """从JWT令牌获取当前用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的认证令牌",
        headers={"WWW-Authenticate": "Bearer"}, # 告诉客户端如何请求认证令牌，标准
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
  
    user = get_user(username)
    if user is None:
        raise credentials_exception
    return user

哈希密码

使用 passlib[bcrypt] 库

python# 密码上下文配置 (使用Bcrypt算法)
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

pythondef verify_password(plain_password: str, hashed_password: str) -> bool:
    """使用Bcrypt验证密码"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """使用Bcrypt生成密码哈希"""
    return pwd_context.hash(password)

中间件

中间件就是一个介于前端和后端路径函数之间的一个函数，它在路径函数之前接收路径函数的请求，可以对路径函数接受到的请求进行处理后再发送给路径函数。

并且后面还会得到路径函数返回的response，并对response进行处理后再返回给前端。

pythonimport time

@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

pythonbase_url = "http://localhost:8000/middleware"

测试中间件

python# 耗时操作
@router.get("/sleep/{time}")
async def expensive_operation(time: int):
    await asyncio.sleep(time)
    return "done"

pythonseconds = 2

response = requests.get(base_url + f"/sleep/{seconds}")
print(response.status_code, response.json())
# 获取headers
print(response.headers["X-Process-Time"])

plaintext    200 done
    2.002873189001548

CORS 跨域资源共享

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

因此，这些都是不同的源：

http://localhost
https://localhost
http://localhost:8080
即使它们都在 localhost 中，但是它们使用不同的协议或者端口，所以它们都是不同的「源」。

可以配置使用不同的origin是否可以访问

pythonfrom fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost",
    "http://localhost:8080",
    # "*" # 通配符，允许所有来源
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True, # 允许跨域请求带上cookie
    allow_methods=["*"], # 允许跨域请求的方法，如：GET、POST、PUT、DELETE等
    allow_headers=["*"], # 允许跨域请求的header， 如：Content-Type、Authorization等
)


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

结语

也算是把基本要用的东西和概念都学完了，剩下的就是实践和深入学习。

目前我按照最佳实践搭了一个项目模版，已经放在了 github。

FastAPI 学习笔记

参数