zl程序教程

您现在的位置是:首页 >  其它

当前栏目

FastAPI 基本使用(一)

基本 FastAPI 使用
2023-09-11 14:21:25 时间

 

目录

​编辑

1、基本介绍 

2、运行方式

3、创建接口步骤

4、自动生成API文档

4.1 交互式API文档

4.2 备用API文档

5、FastApi 执行顺序

6、Python-Pydantic库

6.1 BaseModel模型

6.2 请求体 + 路径参数 + 查询参数

7、Query/Path/Body/Field 参数(额外的校验) 与 字符串验证

8、typing类型注解

8.1 常用类型提示

8.2 Optional 可选类型

8.3 Union 联合类型

8.4 typing List

9、请求示例展示在接口文档中

10 、Cookie,Header参数

11 、响应模型 response_model

12 、响应状态码-使用 status_code 参数来声明

13 、Form表单数据

14 、上传文件-File, UploadFile

15 、处理错误-HTTPException

16、jsonable_encoder() 函数

17 、依赖注入-Depends


1、基本介绍 

        基于Python3.6+版本的、用于构建API现代的、高性能的web框架。FastAPI是建立在Pydantic和Starlette基础上的,Pydantic是一个基于Python类型提示来定义数据验证、序列化和文档的库。Starlette是一种轻量级的ASGI框架/工具包,是构建高性能Asyncio服务的理性选择。

  1. Uvicorn 是一个闪电般快速的 ASGI 服务器,基于 uvloop 和 httptools 构建。
  2. 实现一个基于ASGI(异步服务器网关接口)的最小应用程序接口。
  3. Starlette 负责 web 部分。
  4. Pydantic 负责数据部分。
  5. https://fastapi.tiangolo.com/zh/tutorial/first-steps/

2、运行方式

  1. 运行命令 uvicorn main:app --reload
  2. pycharm运行 (通过 uvicorn 命令行 uvicorn 脚本名:app对象--reload 参数 启动服务)
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

3、创建接口步骤

  1. 导入 FastAPI(from fastapi import FastAPI);
  2. 创建一个 app 实例(app = FastAPI());
  3. 编写一个路径操作装饰器(如 @app.get("/"));
  4. 编写一个路径操作函数(如上面的 def root(): ...);
  5. 定义返回值运行开发服务器(如 uvicorn main:app --reload);

4、自动生成API文档

4.1 交互式API文档

在浏览器中请求 http://127.0.0.1:8000/docs ,显示交互式API文档, 自动交互式 API 文档(由 Swagger UI 提供),如下图:

4.2 备用API文档

在浏览器中请求 http://127.0.0.1:8000/redoc ,显示备用API文档, 备用的自动交互式文档(由 ReDoc 提供),如下图:

 

5、FastApi 执行顺序

按照路径顺序匹配,默认匹配的是第一个带参数的路径

6、Python-Pydantic库

  1. pydantic库是一种常用的用于数据接口schema定义与检查的库。
  2. 通过pydantic库,我们可以更为规范地定义和使用数据接口。
  3. pydantic库的数据定义方式是通过BaseModel类来进行定义的,所有基于pydantic的数据类型本质上都是一个BaseModel类。调用时,我们只需要对其进行实例化即可。
  4. 是一个用来执行数据校验的python库。
from pydantic import BaseModel

# schema基本定义方法
class Person(BaseModel):
    name: str

# 基本的schema实例化方法-直接传入
p = Person(name="ABC123")
print(p.json())

>>> {"name": "ABC123"}

6.1 BaseModel模型

  1. 可以在代码运行时强制执行类型提示,并在数据校验无效时提供友好的错误提示。
  2. 是一个解析库,而不是验证库(简单来说:pydantic保证输出模型的类型和约束)
  3. 所有基于 pydantic 的数据类型本质上都是一个 BaseModel 类
  4. 使用pydantic模型作请求体
  5. 需要安装 pip install pydantic
  6. 使用时需要导入 from pydantic import BaseMode
from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    status: Optional[bool] = None

6.2 请求体 + 路径参数 + 查询参数

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

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

app = FastAPI()

"""
函数参数将依次按如下规则进行识别:
● 如果在路径中也声明了该参数,它将被用作路径参数(例如:item_id)。
● 如果参数属于单一类型(比如 int、float、str、bool 等)它将被解释为查询参数(例如:q)。
● 如果参数的类型被声明为一个 Pydantic 模型,它将被解释为请求体(例如:item)。
"""

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

7、Query/Path/Body/Field 参数(额外的校验) 与 字符串验证

  1. body,Query,Path方法的父类都是直接或者间接的继承FieldInfo,而Field就FieldInfo的实例化,FieldInfo继承了Representation,它们本质上就是Representation类的子类。
  2. 从 fastapi 导入 Query、Path 等对象时,他们实际上是返回特殊类的函数。
  3. 使用Path提取和验证路径参数;使用Query提取和验证请求中的参数;使用Body将参数让客户端由body(默认application/json方式)传入。
  4. Field 可用于提供有关字段和验证的额外信息,如设置必填项和可选,设置最大值和最小值,字符串长度等限制
  5. 参数说明:
    1. Field(None) 是可选字段,不传的时候值默认为None
    2. Field(…) 是设置必填项字段
    3. title 自定义标题,如果没有默认就是字段属性的值
    4. description 定义字段描述内容
    5. gt:大于(greater than)
    6. ge:大于等于(greater than or equal)
    7. lt:小于(less than)
    8. le:小于等于(less than or equal)
from fastapi import FastAPI, Query

# m是可选参数,参数长度2-10,以name开头
@app.get("/update_items/")
# 当使用 Query 且需要声明一个值是必需的时,可以将 ... 用作第一个参数值
# def update_items(m: Optional[str] = Query(..., max_length=10,min_length=2,regex="^name")):
def update_items(m: Optional[str] = Query(None, max_length=10,min_length=2,regex="^name")):
    results = {"items": [{"oneid": "北京"}, {"two": "上海"}]}
    if m:
        results.update({"上海": m})
    return results


"""
声明数值校验:
gt:大于(greater than)
ge:大于等于(greater than or equal)
lt:小于(less than)
le:小于等于(less than or equal)

regex正侧参数的写法:
以 ^ 符号开头
以 $ 符号结尾

使用省略号(...)声明必需参数
"""

# 需求:id大于5才能返回
@app.get("/id")
async def read_id(*, id: int = Query(..., ge=5, ), q_value: str):
    results = {"message": f"{q_value}"}
    if id:
        results.update({"new_value": q_value})
    return results

"""
● 传递 * 作为函数的第一个参数,
● 如果单独出现星号 *,
● 则星号 * 后的参数必须用关键字传入
"""

@app.get("/items/{item_id}")
# 使用 Path 为路径参数声明相同类型的校验和元数据
async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


"""
●嵌入单个请求体参数,期望一个拥有 item 键并在值中包含模型内容的 JSON,
●返回如下:
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}
●可以使用一个特殊的 Body 参数 embed : item: Item = Body(embed=True)
"""

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


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(embed=True)):
    results = {"item_id": item_id, "item": item}
    return results


# 通过get方式在URL路径中接收请求参数
@app.get("/items/{item_id}")
async def read_root1(item_id: int = Path(..., gt=0)):
    return {"item_id": item_id}

# 虽然是通过post方式提交请求,但item_id仍然是在URL中通过?item_id进行请求
@app.post("/items")
async def read_root2(item_id: int = Query(..., ge=1)):
    return {"item_id": item_id}

# post方式提交请求,但item_id仍然是在URL中通过?item_id进行请求
@app.post("/items")
async def read_root2(
	item_id: int = Body(..., ge=1, embed=True),
	item_name: str = Body(None, max_length=20)):
    return {"item_id": item_id, "item_name": item_name}


8、typing类型注解

typing —— 类型注解支持 — Python 3.11.0 文档

8.1 常用类型提示

前两行小写的不需要 import,后面三行都需要通过 typing 模块 import

  • int,long,float: 整型,长整形,浮点型;
  • bool,str: 布尔型,字符串类型;
  • List, Tuple, Dict, Set:列表,元组,字典, 集合;
  • Iterable,Iterator:可迭代类型,迭代器类型;
  • Generator:生成器类型;
# 下面的函数接收与返回的都是字符串,注解方式如下:
def items(name: str) -> str:
    return "hello" + name

print(items("123"))

>>>hello123

8.2 Optional 可选类型

        (声明a :Optional[int] = None) ,参数除了给定的默认值外还可以是None(作用是让编译器识别到该参数有一个类型提示,可以使指定类型,也可以是None,且参数是可选非必传的)。注意 Optional[] 里面只能写一个数据类型。即 Optional[X] 等价于 Union[X, None]

typing —— 类型注解支持 — Python 3.11.0 文档

# 参数end是可选择的参数,有end就返回需求的,没有end返回所有的
@app.get("/params")
async def read_param(start: int = 0, end: Optional[int] = None):
    if end:
        return data[start:end]
    return data[start::]

8.3 Union 联合类型

# Union[int, None] 表示既可以是 int,也可以是 None。没有顺序的说法

python 3.6 及以上版本,需要导入typing中的Union
from typing import Union

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[int, None] = None


python 3.9 及以上版本 ,不需要导入typing中的Union
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: int | None = None


8.4 typing List

  1. typing 模块中导入 List,可以定义类型sellarea : List[str] = [];
  2. typing 的 List、Set、Tuple 都会指定里面参数的数据类型;
  3. FastAPI 会对声明了数据类型的数据进行数据校验,所以会针对序列里面的参数进行数据校验;
  4. 如果校验失败,会报一个友好的错误提示;
  5. 使用时需要导入 from typing import List, Tuple;
from typing import Optional
 
import uvicorn
from fastapi import FastAPI, Body
from typing import List, Tuple
 
app = FastAPI()
 
 
@app.put("/items/{item_id}")
async def update_item(
        list_: List[int] = Body(...),
        tuple_: Tuple[int] = Body(...),
):
    results = {"list_": list_, "tuple_": tuple_}
    return results
 
 
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8080)

9、请求示例展示在接口文档中

使用Config 和 schema_extra 为Pydantic模型声明一个简单的示例

import uvicorn
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel, Field
from typing import Union

class Items(BaseModel):
    name: str
    desc: Optional[str] = None
    price: float
    tax: Optional[float] = None
    
# 接口请求示例展现在接口文档中
"""使用Config 和 schema_extra 为Pydantic模型声明一个简单的示例"""
    class Config:
        schema_extra = {
            "example": {
                "name": "书名",
                "price": 20,
                "decs": "描述信息",
                "tax": 0.5
            }
        }


@app.post("/items1")
async def retrun_item(item: Items):
    results = {"item": item}
    return results


"""
通过工厂函数,增加example参数
注意:传递的额外参数不会添加任何验证,只会添加注释,用于文档的目的
"""

class Item(BaseModel):
    name: str = Field(example="Foo")
    description: Union[str, None] = Field(default=None, example="A very nice Item")
    price: float = Field(example=35.4)
    tax: Union[float, None] = Field(default=None, example=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

if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

 ee2ef1e5ff239ecb9d5660bd54e616ad.png

10 、Cookie,Header参数

  1. Header 是 Path, Query 和 Cookie 的兄弟类型。它也继承自通用的 Param 类;
  2. 从fastapi导入 Query, Path, Header, 或其他时,实际上导入的是返回特定类型的函数;
"""
header的必须有token且token必须是456,没有返回无权限,
cookie必须有一个name,且等于123,否则返回认证失败
"""
from typing import Optional
from fastapi import Cookie, FastAPI,Header
app = FastAPI()
@app.get("/items/")
def read_items(name:  Optional[str] = Cookie(None),
               token: Optional[str] =  Header(None)):
    if token is None or token!='456':
        return '无权限'
    if name is None or name !="123":
        return  "认证失败"

11 、响应模型 response_model

class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: Optional[str] = None

class Userout(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

@app.post("/user/", response_model=Userout)
def create_user(user: UserIn):
    return user

12 、响应状态码-使用 status_code 参数来声明

https://fastapi.tiangolo.com/zh/tutorial/response-status-code/

  1. 在任意的路径操作中使用 status_code 参数来声明用于响应的 HTTP 状态码
  2. status_code 参数接收一个表示 HTTP 状态码的数字
  3. @app.post("/items/", status_code=201)

13 、Form表单数据

  1. 需预先安装 pip install python-multipart;
  2. 使用时需要导入 from fastapi import FastAPI, Form;
  3. Form 是直接继承自 Body 的类;
  4. 表单数据的「媒体类型」编码一般为 application/x-www-form-urlencoded;
import uvicorn
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/",status_code=200)
def login(username: str = Form(...), password: str = Form(...)):
    if password == "123456":
        return {"username": username}
    return "密码错误"
    
# 注册用户,username长度8-16位,password长度6-16位,符合需求返回对应username
@app.post("/register", status_code=200)
async def register(username: str = Form(..., min_length=8, max_length=16, regex='^[a-zA-Z]'),
                   password: str = Form(..., min_length=8, max_length=16, regex='^[0-9]')):
    return {"username": username}


if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

14 、上传文件-File, UploadFile

  1. 需要预先安装 pip install python-multipart;
  2. 使用时需要导入 fromfastapiimportFastAPI,File,UploadFile;
  3. 声明文件体必须使用 File,否则,FastAPI 会把该参数当作查询参数或请求体(JSON)参数;
  4. UploadFile 的属性如下:
    1. filename:上传文件名字符串(str);
    2. UploadFile 支持async 方法;
    3. file: SpooledTemporaryFile( file-like 对象)。其实就是 Python文件,可直接传递给其他预期 file-like 对象的函数或支持库;
    4. content_type:内容类型(MIME 类型 / 媒体类型)字符串(str);
  5. UploadFile 与 bytes 相比有更多优势:
    1. 使用 spooled 文件:存储在内存的文件超出最大上限时,FastAPI 会把文件存入磁盘;
    2. 暴露的 Python SpooledTemporaryFile 对象,可直接传递给其他预期「file-like」对象的库;
    3. 自带 file-like async 接口;
    4. 可获取上传文件的元数据;
    5. 这种方式更适于处理图像、视频、二进制文件等大型文件,好处是不会占用所有内存;
  6. UploadFile 直接继承自 Starlette 的 UploadFile;
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")
def create(file: bytes = File(...)):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

15 、处理错误-HTTPException

  1. 使用时需要导入 from fastapi import FastAPI, HTTPException;
  2. 触发 HTTPException 时,可以用参数 detail 传递任何能转换为 JSON 的值,不仅限于str;
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"test": "浅说测试开发"}
@app.get("/items/{item_id}")
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]}

16、jsonable_encoder() 函数

https://fastapi.tiangolo.com/zh/tutorial/encoder/#__tabbed_1_2

  1. FastAPI 内部用来转换数据的;
  2. 它接收一个对象,比如Pydantic模型,并会返回一个JSON兼容的版本;
  3. 使用时需要导入 from fastapi.encoders import jsonable_encoder;

17 、依赖注入-Depends

  1. from fastapi import Depends, FastAPI;
  2. 不带括号时,调用的是这个函数本身 ,是整个函数体,是一个函数对象,不须等该函数执行完成;
  3. 带括号(参数或者无参),调用的是函数的执行结果,须等该函数执行完成的结果;
  4. 多次使用同一个依赖项:
    1. 如果在同一个路径操作 多次声明了同一个依赖项,例如,多个依赖项共用一个子依赖项,FastAPI 在处理同一请求时,只调用一次该子依赖项;
    2. FastAPI 不会为同一个请求多次调用同一个依赖项,而是把依赖项的返回值进行「缓存」,并把它传递给同一请求中所有需要使用该返回值的「依赖项」;
from fastapi import Depends

# 定义依赖项函数
def com_methods(q: Optional[str] = None, a: int = 0, b: int = 10):
    return {"q": q, "a": a, "b": b}

@app.get("/com_item1")
def com_item1(item1: dict = Depends(com_methods)):
    return item1

@app.post("/com_item2")
def com_item2(item2: dict = Depends(com_methods)):
    return item2


"""
class CommonQueryParams:

类实现依赖注入2中写法:
1.commons: CommonQueryParams = Depends(CommonQueryParams)
2.commons: CommonQueryParams = Depends()
"""

# 全局都需要校验token

from fastapi import  FastAPI,Header, HTTPException,Depends

def verify_token(token: str = Header(...)):
    if token!="asdfghjkl":
        raise HTTPException(status_code=400, detail="Token header invalid")
app = FastAPI(dependencies=[Depends(verify_token)])