FastAPI是什么

FastAPI 是一个用于构建 API 的 web 框架，使用 Python 并基于标准的 Python 类型提示。

与flask相比有什么优势

高性能：得益于uvloop，可达到与 NodeJS 和 Go 并肩的极高性能

简单：fastapi的设计的易于使用和学习。

文档完善：有官方中文文档，详细有条理，非常便于学习。

自动生成api文档：自动生成兼容OpenAPI相关开放标准的api文档

如何学习

1. 最好的学习资源是官方文档，有中文版，非常详细清晰 fastapi.tiangolo.com/zh/learn/[1]
2. 跟着代码亲自输入一遍

下面以 安装、获取请求数据、上传文件、中间件、跨域、后台任务 等常用功能来简单聊聊，同时介绍一个搭配 FastAPI 的高性能异步服务器 uvicorn

安装 FastAPI

安装非常简单，同安装其他依赖没什么区别。

pip install fastapi

不过FastAPI不像flask，会自带一个开发服务器，FastAPI需要额外安装ASGI服务器，官方推荐搭配 uvicorn

安装命令 pip install "uvicorn[standard]"

可选依赖

在使用不同功能时，需要额外安装一些如下依赖。

httpx jinja2 python-multipart itsdangerous pyyaml graphene uvicorn orjson ujson

可以预先通过命令 pip install "fastapi[all]" 来安装以上所有依赖。

一个简单的示例

访问首页/ 返回 hello world

访问 /users/数字根据输入的userID返回问候

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return "Hello World"


@app.get("/users/{user_id}")
def read_item(user_id: int):
    return {"user_id": f'hello, you userID is {user_id}'}

打开 cmd 终端窗口，执行 uvicorn main:app --host 0.0.0.0 --port 8080, 然后浏览器访问 http://127.0.0.1:8080和 http://127.0.0.1:8080/users/123 即可看到返回的响应。

够简单吧。在这个过程中，FastAPI 将会自动执行以下操作：

• 校验 GET 请求的路径中是否含有 user_id 参数。
• 校验 GET 请求中的 user_id 的值是否为 int 类型。
• 如果不是，客户端将会收到清晰有用的错误信息。
• 自动对 JSON 进行转换或转换成 JSON。

指定模板目录和使用模板文件

FastAPI 多用于 API 开发，需要模板的场景不多，但需要时也可以支持。 要使用模板，请先安装jinja2依赖，安装命令 pip install jinja2

在使用前，先创建一个模板文件夹，并在代码中使用Jinja2Templates指定该目录，在需要返回模板响应的函数中指定模板文件名字和绑定的数据。

from fastapi import Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse

app = FastAPI()

templates = Jinja2Templates(directory="templates")

@app.get("/", response_class=HTMLResponse)
def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

如上，以 Jinja2Templates(directory="templates") 指定当前模板存储于当前目录下的templates文件夹内，在返回响应中，指定使用index.html 模板，并绑定模板中可使用的数据requests

指定静态资源文件夹

同模板需要明确指定一样，静态文件夹也需要手动挂载

from fastapi.staticfiles import StaticFiles

app = FastAPI()

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

?

更推荐使用nginx作为前置，作为静态文件服务器，性能更高。

定义路由

创建一个路由：创建url路径装饰器，以及处理函数。

路径指的是 URL 中从第一个 / 起的后半部分，例如/,/users

定义一个绑定到/的GET方法的路由，代码如下

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

这就是一个GET 请求路由，当直接访问域名不加任何路径时，将使用这个处理，例如 http://xxxx.com/

@app.get("/")是装饰器，get指定请求方法为GET， 下方的函数是路径操作函数，用于接收请求信息并返回响应

可以返回一个 dict、list，像 str、int 一样的单个值，或者 Pydantic 模型

如果要捕获针对/的POST请求， 只需要新加一个路由，使用 @app.post('/')装饰器即可。如下

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

?

你也可以使用其他的操作：

@app.post()

@app.put()

@app.delete()

以及更少见的：

@app.options()

@app.head()

@app.patch()

@app.trace()

获取GET请求数据

1. 路径参数：

• 在装饰器中声明路径参数：支持f''格式的路径参数定义 ，例如 @app.get('/user/{user_id}')
• 在请求中获取参数值：定义后可直接在处理函数参数上声明该参数及类型，函数体内直接使用

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

上例中 user_id:int 声明为int型，FastAPI 自动将路径中对应的user_id字符串转为int类型并赋值给参数user_id,函数体中可直接使用

2. 查询参数：

查询参数即url上的查询字符串，如xx.com/search?q=hello&page=1

函数声明中的参数，除了路径参数之外的，FastAPI均将他们当做查询参数，去获取和转为指定类型。

同样上面这个示例，额外增加2个参数，keyword:str,limit:int

@app.get("/items/{user_id}")
    async def read_item(user_id: int,keyword:str,limit:int):
        
        return {"uid": user_id,"keyword":keyword,""}

fastapi将会从url查询字符串中，例如http://xx.com/items/123?keyword=myname&limit=10

获取到函数中定义的同名参数，并转为指定的类型

?

设为可选参数：如果某个参数可以不存在，只需要简单的指定一个默认值None即可，如limit:int=None,当然也可以指定默认值，只需要将None改为想设置的数字，limit:int=10

Pydantic 数据模型

在使用 POST 请求获取请求体之前，需要提到 Pydantic 数据模型，这是FastAPI能简单而高效的一个重要原因.

?

Pydantic 是 Python 中使用最广泛的数据验证库。使用它可以非常方便的指定请求和返回的数据，全程自动json化，并自动校验正确性，极大的提高开发效率。

并且底层使用RUST开发，性能很高。

要使用它：

1. 从 pydantic 中导入 BaseModel
2. 创建数据模型类并继承 BaseModel

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

FastAPI将自动将该数据模型转为JSON 对象

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

在POST请求中就可以直接使用它了，如示例

@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

在这个请求中，检测到 items 是一个数据模型类型，FastAPI就会自动从post请求体中获取模型声明的对应参数，并赋值给 item，在函数体中则可以直接使用它。比自行处理body要方便很多。

获取POST请求体

JSON body请求体/application/json

1. 声明请求体参数

使用与声明查询参数相同的方式声明请求体：

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

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

如上，声明了一个请求体item,类型是Item,FastAPI将自动从请求体中获取并转为Item类型的数据赋值给item

2. 获取请求体中的数据

在函数中中即可直接使用 item.name item.price获取请求参数值

?

上述声明后，Fastapi将自动进行以下操作

以 JSON 形式读取请求体

（在必要时）把请求体转换为对应的类型

校验数据：

数据无效时返回错误信息，并指出错误数据的确切位置和内容

把接收的数据赋值给参数 item

把函数中请求体参数的类型声明为 Item，还能获得代码补全等编辑器支持

普通表单数据获取

接收的不是 JSON，而是表单字段时，要使用 Form 模块

先导入from fastapi import FastAPI,Form

声明方法依然和查询参数声明一样，在函数上以参数形式定义，不同的是必须指定默认值Form().

from fastapi import FastAPI, Form

app = FastAPI()


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

如果没有指定，FastAPI只会将它当做查询参数来处理

获取上传文件

?

因为上传文件以「表单数据」形式发送。所以接收上传文件，要预先安装 pip install python-multipart

定义文件参数时使用 UploadFile 模块

先导入 from fastapi import FastAPI, File, UploadFile

声明方式一如既往，在函数中以参数定义，指定数据类型类为UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

剩下的复杂操作FastAPI会自动处理。

上述代码中的 file即是获取到的上传文件，它是一个UploadFile对象，属性如下

UploadFile 的属性如下：

filename：上传文件名字符串（str），例如， myimage.jpg；

content_type：内容类型（MIME 类型 / 媒体类型）字符串（str），例如，image/jpeg；

file： file-like[2] 对象。其实就是 Python文件，可直接传递给其他预期 file-like 对象的函数或支持库。

UploadFile 支持以下 async 方法

write(data)：把 data （str 或 bytes）写入文件；

read(size)：按指定数量的字节或字符（size (int)）读取文件内容；

seek(offset)：移动至文件 offset （int）字节处的位置；

-   例如，`await myfile.seek(0)` 移动到文件开头；
-   执行 `await myfile.read()` 后，需再次读取已读取内容时，这种方法特别好用；

close()：关闭文件。

因为上述方法都是 async 方法，要搭配「await」使用。

例如读取contents = await myfile.read()

如果你的函数没有使用async声明，则不可使用await，可使用以下方法替代

contents = myfile.file.read()

设置和返回cookie

1. 从请求中获取cookie

和定义路径参数、查询参数类似，只需要在函数中定义cookie参数，即可直接使用。类似

from typing import Union

from fastapi import Cookie, FastAPI
from typing_extensions import Annotated

app = FastAPI()


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

当然形式有所变化，需要先从typing_extensions中导入Annotated

ads_id: Annotated[Union[str, None], Cookie()] = None 表示从cookie中取出名为ads_id的cookie的值

?

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

?

Annotated 是Python中的一个装饰器（Decorator），它的作用是为函数、方法、类等对象添加额外的注释信息。这些注释信息可以包括参数和返回值的类型、文档字符串的内容、函数的调用示例等。使用 Annotated 能够帮助开发者更好地了解代码的用途和使用方式，提高代码的可读性和可维护性。

2. 响应Cookies

要在响应中返回cookie，需要定义一个类型为 Response的参数，这样你就可以在这个临时响应对象中设置cookie

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return {"message": "Come to the dark side, we have cookies"}

中间件

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

1. 创建中间件

要创建中间件可以在函数的顶部使用装饰器,例如 @app.middleware("http").

中间件参数接收如下参数:

• request
• 一个函数 call_next 它将接收 request 作为参数.
• 这个函数将 request 传递给相应的 路径操作.
• 然后它将返回由相应的路径操作生成的 response.
• 然后你可以在返回 response 前进一步修改它.

@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

跨域处理

在 FastAPI 应用中使用 CORSMiddleware中间件 来配置跨域。

• 导入 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"}

CORS 预检请求

在复杂的请求时，需要有一个OPTIONS预检请求，这是带有 Origin 和 Access-Control-Request-Method 请求头的 OPTIONS 请求。

在这种情况下，中间件将拦截传入的请求并进行响应，出于提供信息的目的返回一个使用了适当的 CORS headers 的 200 或 400 响应。

简单后台任务

所谓后台任务，其实就是将相关数据交给另一个函数，然后立即返回，无需等待另一个函数的返回结果，适合于需要较长时间运行的任务，比如发送短信、邮件等。

也适合在异步处理函数中调用同步函数的情况。

当然如果你需要立即获得响应，那么不适合使用后台任务

后台任务使用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"}

实现步骤：

1. 首先创建一个函数，实现实际执行任务的功能
2. 在请求中注入 BackgroundTasks 然后调用 add_task, 添加任务和数据

.add_task() 接收以下参数：

• 在后台运行的任务函数(write_notification)。
• 应按顺序传递给任务函数的任意参数序列(email)。
• 应传递给任务函数的任意关键字参数(message="some notification")。

uvicorn 服务器

fastapi不自带web服务器，官方推荐 uvicorn 。这是一个高性能 ASGI 服务器

pip install "uvicorn[standard]"

安装后即可在命令行中启动服务 uvicorn main:app --host 0.0.0.0 --port 80

当然也可以直接从代码中启动

import uvicorn

async def app(scope, receive, send):
    ...

if __name__ == "__main__":
    uvicorn.run("main:app", port=5000, log_level="info")

原文：https://juejin.cn/post/7396932646590627852

作者：mortimer

#记录我的2024#

Reference

[1] https://fastapi.tiangolo.com/zh/learn/: https://fastapi.tiangolo.com/zh/learn/

[2] https://docs.python.org/zh-cn/3/glossary.html#term-file-like-object: https://docs.python.org/zh-cn/3/glossary.html#term-file-like-object