介绍

FastAPI 是一个用于构建 Web API 的 Python 框架.

安装

1	`pip install "fastapi[all]"`

(这里其实包括 uvicorn 的安装了)

另外要安装一个 ASGI (Asynchronous Server Gateway Interface, 负责在网络请求和 Python 应用程序之间进行通信) 服务器, 如 Uvicorn:

1	`pip install "uvicorn[standard]"`

示例

创建 main.py 并写入:

# 导入 FastAPI
from fastapi import FastAPI

# 创建一个叫 app 的实例
app = FastAPI()

# 创建一个路径修饰器, 表明用 GET 访问 / 时调用后面的函数
@app.get("/")
async def read_root():
    return {"Hello": "World"}

运行:

1	`uvicorn main:app --reload`

main:app, 前者指运行 main.py 文件, 后者指定文件中创建的 FastAPI 实例名称
--reload, 指启用自动重载，当代码更改时，服务器会自动重启

用浏览器访问 http://127.0.0.1:8000/items/5?q=somequery 来测试:

调试

可以访问 http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc 来调试 API.

语法

路径参数

可以理解为变量插值 (实际上是用来捕获), 可以使用 {var} 的语法:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

路径顺序

注意路径的匹配是按照编写顺序的:

from fastapi import FastAPI

app = FastAPI()

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

若反过来行为就会不同. (反过来的情况, 比如访问 /users/me 这里的 me 会被 {user_id} 捕获)

预定义值

需要借助 Enum 类来实现:

from enum import Enum

from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    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"}

    return {"model_name": model_name, "message": "Have some residuals"}

通过继承 str 类, 可以把该枚举值定义为 “字符串”.

路径参数

如:

from fastapi import FastAPI

app = FastAPI()

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

{file_path:path}, 这里的 file_path 就是路径参数, :path 是修饰, 其表明 file_path 捕获任意长度的路径, 如 /files/a/b/c, 那么 file_path 的值为 a/b/c

查询参数

即:

1	`http://127.0.0.1:8000/items/?skip=0&limit=10`

这里的 skip 和 limit 就是查询参数.

体现在 FastAPI 中:

from fastapi import FastAPI

app = FastAPI()

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]

这里同时给 skip 和 limit 设置了默认值.

类型注解和可选参数

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

这里的 str | None 就是类型注解, 表明其可以是 str 类型, 也可以是 None 类型, 而当用户没有提供 q 变量时, 其就为 None 类型且默认值为 None.

必选参数

不给参数指定默认值, 那它就为必选参数:

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    item = {"item_id": item_id, "needy": needy}
    return item

这里的 needy 就是必选参数, 若访问:

1	`http://127.0.0.1:8000/items/foo-item`

则会返回错误响应. 正常应为:

1	`http://127.0.0.1:8000/items/foo-item?needy=sooooneedy`

请求体

借助 Pydantic 库 (一个用于数据验证和文档化的 python 库), 一般用其中的 BaseModel 类.

可以转换为字典, JSON 序列化等:

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

这里实际上是返回:

{
    "name": "Foo",
    "description": "An optional description",
    "price": 45.2,
    "tax": 3.5
}

即一个 Python 字典 (也是 JSON 对象).

这里的 item 就是请求体.

若要以字典访问 BaseModel 类实例, 需使用 dict() 方法如:

from fastapi import FastAPI
from pydantic import BaseModel

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

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

给查询参数添加约束

比如, 对于一个查询参数而言, 不能超过 50 个字符的长度, 可以写为:

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(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

设置查询参数的默认值为 Query.

Query 函数的第一个参数用于设置默认值 (比如原来的 None 就写在这里了), 而余下参数用于设置约束.

比如:

1	`q: Union[str, None] = Query(default=None)`

等价于:

1	`q: str = None`

添加更多约束的示例:

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
    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

用 Query 接收一组值

from typing import List, Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):
    query_items = {"q": q}
    return query_items

添加元数据

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    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 result

这里的;

alias
title
description
deprecated

就是元数据

给路径参数添加约束和元数据

from typing import Annotated

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    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

数值校验示例

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

有多个参数的请求体

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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

其期望的请求体为:

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

向请求体添加一个键

也就是期望请求的 JSON 中多一个键值对, 借助 fastapi 提供的 Body():

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


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, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

其期望:

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

也可以给 Body() 传递约束和元数据.

模型内部进行约束和元数据设置

使用 Pydantic 中的 Field:

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

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

添加额外文档信息

给 Field, Body, Path, Query 添加 example 参数即可:

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


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

给查询参数添加 Cookie() 的类型注释就行:

from typing import Annotated

from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}

给查询参数添加 Header() 的类型注释就行:

from typing import Annotated

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}

处理响应类型

即 response_model, 是 get, post 等修饰器方法的参数.

用来将输出数据转换为指定类型以及数据校验等.

如:

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

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

使用多个模型

主要借助继承以及模型间的转换:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

class UserIn(UserBase):
    password: str

class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: str

def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

借助 Union 使用多个模型中的某个:

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class BaseItem(BaseModel):
    description: str
    type: str

class CarItem(BaseItem):
    type: str = "car"

class PlaneItem(BaseItem):
    type: str = "plane"
    size: int

items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

指定响应状态码

使用 status_code 参数:

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

接收表单数据

使用 Form() 函数:

from fastapi import FastAPI, Form

app = FastAPI()


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

接收文件

需要先安装 python-multipart:

1	`pip install python-multipart`

使用 File() 函数或 UploadFile 类.

示例:

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.post("/files/")
async def create_files(
    files: list[bytes] = File(description="Multiple files as bytes"),
):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(
    files: list[UploadFile] = File(description="Multiple files as UploadFile"),
):
    return {"filenames": [file.filename for file in files]}


@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)

接收 PDF 文件的完整示例

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
import shutil
import os
from pathlib import Path

app = FastAPI()

# 指定存储目录
UPLOAD_DIRECTORY = "./uploads"
os.makedirs(UPLOAD_DIRECTORY, exist_ok=True)

@app.get("/", response_class=HTMLResponse)
async def main():
    return """
    <html>
        <head>
            <title>Upload PDF</title>
        </head>
        <body>
            <h1>Upload a PDF file</h1>
            <form action="/uploadfile/" method="post" enctype="multipart/form-data">
                <input type="file" name="file" accept=".pdf" required>
                <button type="submit">Upload</button>
            </form>
        </body>
    </html>
    """

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    if file.filename != None:
        if not Path(file.filename).suffix == '.pdf':
            return {"error": "Only PDF files are allowed."}

        file_path = os.path.join(UPLOAD_DIRECTORY, file.filename)
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)

        return {"filename": file.filename, "filepath": file_path}
    else:
        raise Exception("File name is None")

错误处理

使用 HTTPException 如:

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.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]}

注意这里用 raise 来触发 HTTPException.

定义异常处理器

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


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

路径修饰

给要访问的路径添加一些数据, 如 summary, description, response_description 参数, 以及函数多行注释:

from typing import Set, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    response_description="The created item",
)
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

依赖项

就是提前运行一个函数, 返回一些依赖数据.

函数作为依赖项

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

注意这里的 Dependsh 指接受一个参数, 且参数需要是一个可调用对象.

类作为依赖项

类也是可调用对象:

from fastapi import Depends, FastAPI

app = FastAPI()

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

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

可以把:

1
2
3

...
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
...

简写为:

1
2
3

...
async def read_items(commons: CommonQueryParams = Depends()):
...

不使用依赖项的返回值

当只需要执行依赖项而不需要其返回值时, 在 “路径操作修饰器” 中用 dependencies 指定一个 list:

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

定义全局依赖项

比如让每一个路径操作修饰器都会触发几个依赖项函数:

from fastapi import Depends, FastAPI, Header, HTTPException

async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

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

@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]

@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

安全验证

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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

这里的 tokenUrl 指定的是相对访问路径, 如这里在 /items/ 下, 因此该为 /items/token.

表单会将 username 和 password 发送到该路径, API 在验证之后会响应一个令牌, 之后若要验证身份, 只需要发送值为 Bearer+令牌 的 Authorization 请求头.

中间件

即一个可以在 “请求” 被 “路径操作” 处理前运行, 或在 “响应” 返回前运行的函数.

需要使用 @app.middleware("http) 修饰器来创建. 如:

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.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

call_next 会把 request 传递给相应的路径操作函数处理以获取 response

CORS

在 FastAPI 中用 CORSMiddleware 来配置, 一般步骤为:

创建一个允许的源列表
将其作为中间件添加到 FastAPI 应用中

创建列表示例为:

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

SQL 数据库

ORMs

ORM (Object-Relational Mapping) 是一种编程技术, 它允许你使用面向对象的方式来操作数据库, 而不需要直接编写 SQL 语句.

ORM 的核心思想是将数据库表映射为对象(称为模型或实体), 每个表对应一个类, 每个行对应一个对象实例, 每一列对于一个类成员. 开发者可以使用面向对象的方式来创建, 读取, 更新和删除数据, 而不需要直接与数据库进行交互.

在 Python 中一般使用 SQLAlchemy 包的 ORM 实现. 安装为:

1	`pip install sqlalchemy`

后台任务

指返回响应之后执行的任务.

使用 BackgroundTasks 如:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

步骤为:

创建一个任务函数
将任务函数添加到后台

设置元数据

给 FastAPI 创建的应用实例添加元数据, 如:

from fastapi import FastAPI

description = """
ChimichangApp API helps you do awesome stuff. 🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""

app = FastAPI(
    title="ChimichangApp",
    description=description,
    summary="Deadpool's favorite app. Nuff said.",
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)


@app.get("/items/")
async def read_items():
    return [{"name": "Katana"}]

静态文件

即提供静态文件的响应, 像文件服务器, 使用 StaticFiles 如:

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

把 StaticFiles() 指定的目录挂载到 /static 这个 url 下.

测试

主要借助 pytest 运行测试文件. 安装为:

1	`pip install pytest`

使用 TestClient 测试, 需要先安装 httpx:

1	`pip install httpx`

示例:

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


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


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

将 FastAPI 应用传递给 TestClient() 以创建一个 TestClient 对象.

之后直接运行:

pytest

即可.

分离测试

若文件结构为:

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

则测试文件 test_main.py 文件内容为:

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

命令行工具

在用:

1	`pip install fastapi`

安装 FastAPI 库之后, 会提供 fastapi 命令, 可用于部署, 运行和管理 FastAPI 项目.

开发模式运行

1	`fastapi dev`

此时当你更改代码时, 它会自动重新加载服务器.

此时默认监听 127.0.0.1.

生产模式运行

1	`fastapi run`

此时不会自动重载, 且监听 0.0.0.0.

Python

FastAPI-基本使用

http://example.com/2024/08/19/FastAPI-基本使用/

作者

Jie

发布于

2024年8月19日

许可协议

Python-类型提示上一篇

Amis-基本使用下一篇

FastAPI-基本使用

介绍

安装

示例

调试

语法

路径参数

路径顺序

预定义值

路径参数

查询参数

类型注解和可选参数

必选参数

请求体

给查询参数添加约束

用 Query 接收一组值

添加元数据

给路径参数添加约束和元数据

数值校验示例

有多个参数的请求体

向请求体添加一个键

模型内部进行约束和元数据设置

添加额外文档信息

声明 Cookie 参数

声明 Header 参数

处理响应类型

使用多个模型

指定响应状态码

接收表单数据

接收文件

接收 PDF 文件的完整示例

错误处理

定义异常处理器

路径修饰

依赖项

函数作为依赖项

类作为依赖项

不使用依赖项的返回值

定义全局依赖项

安全验证

中间件

CORS

SQL 数据库

ORMs

后台任务

设置元数据

静态文件

测试

分离测试

命令行工具

开发模式运行

生产模式运行