DW FastAPI Task01

01-路径参数

请确保自己的python版本为3.10+，下面代码运行在ipynb文件中

py中，async和await是用于处理异步编程的关键字，可以重复利用资源处理任务，实现非阻塞的异步操作。

  • 在函数定义前加上async，那么这个函数就变成了异步函数，可以在执行时暂停，然后去执行其他任务。
  • 在调用某个函数前加上await关键字，当前函数将会暂停执行，知道被调用的异步函数执行完毕。

第一个FastAPI程序

!pip install fastapi uvicorn

import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
  #此函数被关键词async修饰，表示此函数为异步函数，可暂停
    return {"message": "Hello World"}

if __name__ == '__main__':
  #配置你的地址和端口号
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    # 异步调用
    await server.serve()
    # 或：uvicorn.run(app,host='0.0.0.0',port=8000),notebook中不能运行run

页面将显示函数返回的内容。

路径参数

import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
  # int类型的item_id
    return {"item_id": item_id}

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    # 异步调用函数，在url中输入item_id
    await server.serve()

预设值

创建一个Enum枚举类，提前确定部分有效参数是可以被接受处理的，

import uvicorn
from fastapi import FastAPI
from enum import Enum

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

app = FastAPI()

# get请求，从我们输入的url中获取model_name值
@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"}

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

包含文件路径参数

当我们需要获取文件时，url参数需要动态的拼接变化

import uvicorn
from fastapi import FastAPI
app = FastAPI()
# 这里的file_path为文件地址，:path表示该参数匹配任意的路径
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

当你输入的文件地址为home/me/readme.md时，file_path将会变化为home/me/readme.md

默认查询参数

对于如下代码

import uvicorn
from fastapi import FastAPI

app = FastAPI()
# 定义一个列表
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


# 定义一个items接口
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
  # 切片语法，表示从skip开始，取到skip+limit-1的元素
    return fake_items_db[skip : skip + limit]

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

在read_item函数中存在两个已经设置好值的参数，类型为int，当我们输入url中并不包含两个参数的值时，会默认使用设置好的。

我们也可以使用/items/?skip=1&limit=2来获取第二和第三个数组的值。

必须查询参数

参考上一节代码，我们在参数列表中去除掉我们设置的默认值即可，表示此参数是必须要传入的。

async def read_user_item(item_id: str, needy: str):

可选查询参数

参考上一节代码，我们在参数列表中，设置参数为None即可表示该参数是可选传入的参数。

async def read_item(item_id: str, q: Union[str, None] = None):

组合使用

import uvicorn
from typing import Union
from fastapi import FastAPI
app = FastAPI()
# 有四个参数需要我们传入，其中：user_id和item_id是必须的，因为没有设置默认值，并且不为None；q为可选传入的  参数，short为存在默认值的参数，可以传入值也可以不传入
@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

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

浏览器输入 http://127.0.0.1:8000/users/3/items/4?q=foo&short=yes

会看到 {“item_id”:“4”,“owner_id”:3,“q”:“foo”}

浏览器输入 http://127.0.0.1:8000/users/3/items/4?q=foo

会看到 {“item_id”:“4”,“owner_id”:3,“q”:“foo”,“description”:“This is an amazing item that has a long description”}

路径参数校验

import uvicorn
from typing import Annotated
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get")]
):
    results = {"item_id": item_id}
    return results

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

此处我们导入了新的包：Path和Annotated

在函数声明阶段，我们在参数列表中使用Annotated来检查用户输入的item_id的值是否为int类型

也可以在path参数后面加上第三个参数：

可选的声明数值校验：

  • gt：大于（greater tha n）
  • ge：大于等于（greater than or equ al）
  • lt：小于（less t han）
  • le：小于等于（less than or equal）
import uvicorn
from typing import Annotated
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[float, Path(title="The ID of the item to get", ge=5.5)]
  # 表示用户输入的参数必须要大于等于5.5
):
    results = {"item_id": item_id}
    return results

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

查询参数校验

在定义函数参数列表阶段，设置某一参数为str|None = None表示该参数为str类型，可不选，默认值为None。

async def read_items(q: str | None = None):

对于q的长度，可以使用如下方法进行限制和判断

# 导入新的包Query
from fastapi import Query
app = FastAPI()
@app.get("/items/")
async def read_items(q: str | None = Query(default=None, max_length=5)):
  # 可以看到，我们使用Query函数作为q的默认值，默认为None，并且，q最大的长度为5
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

在浏览器中输入 http://127.0.0.1:8000/items/?q=testdata

会看到 {“detail”:[{“type”:“string_too_long”,“loc”:[“query”,“q”],“msg”:“String should have at most 5 characters”,“input”:“testdata”,“ctx”:{“max_length”:5},“url”:“https://errors.pydantic.dev/2.5/v/string_too_long”}]}

  • 同样的，使用min_length来限制参数的最小长度，比如用户输入电话号码或者身份证号的时候。
  • 亦或者，使用pattern来加入正则表达式，来处理参数中你不要的字符
import uvicorn
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(q: str | None = Query(default=None, min_length=3, max_length=5, pattern="^test.?$")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

if __name__ == '__main__':
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)
    await server.serve()

这个指定的正则表达式通过以下规则检查接收到的参数值：

^：字符开头，符号之前没有字符

test: 值精确地等于test

.?：零个或一个任意字符。

$: 字符结尾，符号之后没有更多字符。

根据上述规则，test、tests、testa、testb、testc、testd都是允许的字符。 testab、testabc、text则不是被允许的。

当你不需要声明该参数是可选的，而是必须的参数，请去除掉None即可

async def read_items(q: str = Query(min_length=3, max_length=5, pattern="^test.?$")):

使用省略号声明必须参数

该方法可显示的声明此参数是必须的

async def read_items(q: str = Query(default=..., min_length=3)):

在default中设置省略号，表示此参数是必须的，但你也可以直接去除掉default。

此外还有title参数和description参数，作为函数的介绍和标题。

# fastapi # python # 人工智能
Logo
天启AI社区

GitCode 天启AI是一款由 GitCode 团队打造的智能助手，基于先进的LLM（大语言模型）与多智能体 Agent 技术构建，致力于为用户提供高效、智能、多模态的创作与开发支持。它不仅支持自然语言对话，还具备处理文件、生成 PPT、撰写分析报告、开发 Web 应用等多项能力，真正做到“一句话，让 Al帮你完成复杂任务”。

更多推荐

cover

【感知机】感知机(perceptron)学习算法例题及详解

avatar 天启AI社区
cover

AI云原生：数智化时代AI Agent加速落地的技术基石与实践指南（THS）

avatar 天启AI社区
cover

【AI论文】设计实验室：通过迭代检测与修正来设计幻灯片

avatar 天启AI社区