python之FastAPI框架

大家一起进步，感谢大家的支持和关注

fastapi

前言

fastapi框架

1.安装相关模块

2.快速上手

3.路径操作

3.1.include_router

4.请求参数和响应

4.1.路径参数

4.2、查询参数（请求参数）

4.3.请求体数据

4.4、form表单数据

4.5.文件上传

4.6.请求静态文件

4.7.响应模型相关参数

4.7.1. response_model

4.7.2. response_model_exclude_unset

4.7.3. response_model

4.8. request对象

5.jinja2模板

5.1 jinja2 的变量

5.2 jinja2 的过滤器(内置函数)

5.3 jinja2 的条件和循环

6.ORM操作

7.中间件

8.cors

总结

前言

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

fastapi框架

1.安装相关模块

Starlette 负责 web 部分(Asyncio)
Pydantic 负责数据部分(类型提示)
pip install fastapi
pip uvicorn

---

2.快速上手

from fastapi import FastAPI  # FastAPI 是一个为你的 API 提供了所有功能的 Python 类。

app = FastAPI()  # 这个实例将是创建你所有 API 的主要交互对象。这个 app 同样在如下命令中被 uvicorn 所引用


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


# 方式一：终端输入命令启动 
# main是程序的名称
# app 是实例化的名称
# uvicorn main:app --reload


# 方式二
if __name__ == '__main__':
    # 生产环境可以使用[Uvicorn],上线环境用ASGI 服务器
    import uvicorn

    print(__name__)
    uvicorn.run("run:app", host="127.0.0.1", port=8080, reload=True)



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

访问

交互式文档，不用在使用其他测试工具也可以
ip+port\docs

---

3.路径操作

fastapi遵循restfull规范

支持的请求方式
@app.get()
@app.post()
@app.put()
@app.patch()
@app.delete()
@app.options()
@app.head()
@app.trace()

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/get")
def get_test():
    return {"method": "get方法"}


@app.post("/post")
def post_test():
    return {"method": "post方法"}


@app.put("/put")
def put_test():
    return {"method": "put方法"}


@app.delete("/delete")
def delete_test():
    return {"method": "delete方法"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("run:app", host="127.0.0.1", port=8081,  reload=True)

3.1.include_router

创建一个apps文件夹，专门方各种接口的文件

编写入口文件来调用
访问

---

4.请求参数和响应

4.1.路径参数

基本用法
以使用与 Python 格式化字符串相同的语法来声明路径"参数"或"变量"：

有类型的路径参数
后续在浏览器中输入将校验数据类型

顺序
路径相同的，不是动态数据的放前面，不然会覆盖，只用后面的路径对应的视图函数

4.2、查询参数（请求参数）

路径函数中声明不属于路径参数的其他函数参数时，它们将被自动解释为"查询字符串"参数，就是 url? 之后用&分割的 key-value 键值对。
http://127.0.0.1/xx/xxx/?key=value&key2=value2&key3=value3
type hints `主要是要指示函数的输入和输出的数据类型，数据类型在typing 包中，基本类型有str list dict等等，
Union 是当有多种可能的数据类型时使用，比如函数有可能根据不同情况有时返回str或返回list，那么就可以写成Union[list, str]
Optional 是Union的一个简化， 当 数据类型中有可能是None时，比如有可能是str也有可能是None，则Optional[str], 相当于Union[str, None]

from typing import Union

from fastapi import FastAPI
import uvicorn

app = FastAPI()
app.debug = True


@app.get("/jobs/{kd}", tags=["查询参数"])
async def search_jobs(kd: str, city: Union[str, None] = None, salary: Union[str, None] = None):  # 有默认值即可选，否则必选
    '''
    :func : 查询函数
    :param kd:  声明了类型，普通参数
    :param city: 声明了类型，默认参数为None，可选
    :param salary: 声明了类型，默认参数为None，可选
    :return: 返回字典类型
    '''
    if city or salary:
        return {"kd": kd, "city": city, "salary": salary}
    return {"kd": kd}


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

async 异步关键词，后面再说

4.3.请求体数据

当你需要将数据从客户端（例如浏览器）发送给 API 时，你将其作为「请求体」发送。请求体是客户端发送给 API 的数据。响应体是 API 发送给客户端的数据。
FastAPI 基于 Pydantic ，Pydantic 主要用来做类型强制检查（校验数据）。不符合类型要求就会抛出异常。
当一个模型属性具有默认值时，它不是必需的。否则它是一个必需属性。将默认值设为 None 可使其成为可选属性。

FastAPI 会自动将定义的模型类转化为JSON Schema，Schema 成为 OpenAPI 生成模式的一部分，并显示在 API 交互文档中，查看 API 交互文档如下，该接口将接收application/json类型的参数。

FastAPI 支持同时定义 Path 参数、Query 参数和请求体参数，FastAPI 将会正确识别并获取数据。

参数在 url 中也声明了，它将被解释为 path 参数

参数是单一类型（例如int、float、str、bool等），它将被解释为 query 参数

参数类型为继承 Pydantic 模块的BaseModel类的数据模型类，则它将被解释为请求体参数
pip install pydantic

from typing import Union, List, Optional

from fastapi import FastAPI
from pydantic import BaseModel, Field, ValidationError, validator
import uvicorn
from datetime import date


# 自定义封装数据类型
class Address(BaseModel):
    province: str
    city: str


# 封装数据
class User(BaseModel):
    name: str = "ROOT"
    age: int = Field(default=0, lt=100, gt=0)
    birth: Optional[date] = None
    friends: List[int] = []
    description: Union[str, None] = None

    address: Union[Address, None] = None  # 类型嵌套
    address1: Optional[Address] = None  # 类型嵌套

    @validator('name')
    def name_must_alpha(cls, v):
        assert v.isalpha(), 'name must be alpha'
        return v


class Data(BaseModel):  # 类型嵌套
    users: List[User]


app = FastAPI()
app.debug = True

@app.post("/user", tags=["请求体"])
async def user(user: User):
    print(user, type(user))
    print(user.name, user.birth)
    print(user.dict())
    return user


@app.post("/data")
async def data(data: Data):  # 数据统一封装到dat类型
    return data


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

4.4、form表单数据

在 OAuth2 规范的一种使用方式（密码流）中，需要将用户名、密码作为表单字段发送，而不是 JSON。

FastAPI 可以使用Form组件来接收表单数据，需要先使用pip install python-multipart命令进行安装

# 导入Form
from fastapi import FastAPI, Form
import uvicorn

app = FastAPI()


@app.post("/login", tags=["form表单", "form表单1"])
def login(username: str = Form(..., max_length=16, min_length=8, regex='[a-zA-Z]'),
          password: str = Form(..., max_length=16, min_length=8, regex='[0-9]')):
    print(f"username:{username},password:{password}")
    return {"username": username}


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

4.5.文件上传

import uvicorn
from typing import List
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

# file: bytes = File()：适合小文件上传
@app.post("/files/",tags=["file小文件上传"])
async def create_file(file: bytes = File()):
    print("file:", file)
    return {"file_size": len(file)}


@app.post("/multiFiles/",tags=["file多文件上传"])
async def create_files(files: List[bytes] = File()):
    return {"file_sizes": [len(file) for file in files]}


# file: UploadFile：适合大文件上传
@app.post("/uploadFile/",tags=["UploadFile大文件上传"])
async def create_upload_file(file: UploadFile):
    with open(f"{file.filename}", 'wb') as f:
        for chunk in iter(lambda: file.file.read(1024), b''):
            f.write(chunk)

    return {"filename": file.filename}


@app.post("/multiUploadFiles/",tags=["UploadFile多大文件上传"])
async def create_upload_files(files: List[UploadFile]):
    return {"filenames": [file.filename for file in files]}


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

4.6.请求静态文件

在 Web 开发中，需要请求很多静态资源文件（不是由服务器生成的文件），如 css/js 和图片文件等。

# 导入静态相关包
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 设置静态目录文件
app.mount("/static",StaticFiles(directory="static"))

app.debug=True
if __name__ == '__main__':
    uvicorn.run("run:app", host="127.0.0.1", port=8080, reload=True)

访问static下的静态资源

4.7.响应模型相关参数

4.7.1. response_model

# 路径操作
@app.post("/items/", response_model=Item)
# 路径函数
async def create_item(item: Item):
    ...
# response_model 是路径操作的参数，并不是路径函数的参数

以前的return是自定义的响应数据类型
FastAPI将使用response_model进行以下操作

将输出数据转换为response_model中声明的数据类型。

验证数据结构和类型

将输出数据限制为该model定义的

添加到OpenAPI中

在自动文档系统中使用。

你可以在任意的路径操作中使用 response_model 参数来声明用于响应的模型

import uvicorn
from pydantic import BaseModel, EmailStr
from typing import Union
from fastapi import FastAPI

app = FastAPI()



class UserIn(BaseModel):
    """输入模型"""
    username: str = "蜡笔小新"
    password: str = "1234"
    email: EmailStr = "user@example.com"
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    """输出模型"""
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


# response_model 返回响应模型
@app.post("/user/", response_model=UserOut,tags=["响应体"])
async def create_user(user: UserIn): # UserIn 输入模型
    # 返回值定以的模型类型,会自动过滤不需要的字段 
    return user


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

4.7.2. response_model_exclude_unset

我们学到了如何用response_model控制响应体结构，但是如果它们实际上没有存储，则可能要从结果中忽略它们。例如，如果model在NoSQL数据库中具有很多可选属性，但是不想发送很长的JSON响应，其中包含默认值。

import uvicorn
from pydantic import BaseModel, EmailStr
from typing import Union,List
from fastapi import FastAPI

app = FastAPI()


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


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

# 设置为False会显示response_model返回模型的全部字段，默认为False
@app.get("/items/{item_id}", response_model=Item, 
response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]



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

4.7.3. response_model

除了response_model_exclude_unset以外，还有response_model_exclude_defaults和response_model_exclude_none，我们可以很直观的了解到他们的意思，不返回是默认值的字段和不返回是None的字段。
见名知义

# response_model_exclude
@app.get("/items/{item_id1}", response_model=Item, response_model_exclude={"description","name", "price"}, )
async def read_item(item_id1: str):
    return items[item_id1]


# response_model_include
@app.get("/items/{item_id}", response_model=Item, response_model_include={"name", "price"}, )
async def read_item(item_id: str):
    return items[item_id]

4.8. request对象

非常重要——>理解为有很多网络中的数据都封装在里面

希望能直接访问Request对象。例如我们在路径操作函数中想获取客户端的IP地址，需要在函数中声明Request类型的参数，FastAPI 就会自动传递 Request 对象给这个参数，我们就可以获取到 Request 对象及其属性信息，例如 header、url、cookie、session 等。

import uvicorn
from fastapi import FastAPI
# 导包
from fastapi import Request

app = FastAPI()


# 使用时传到函数中
@app.get("/items")
async def items(request: Request):
    return {
        "请求URL：": request.url,
        "请求ip：": request.client.host,
        "请求宿主：": request.headers.get("user-agent"),
        "cookies": request.cookies,
    }


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

---

5.jinja2模板

模板简单来说就是⼀个其中包涵占位变量表⽰动态的部分的⽂件，模板⽂件在经过动态赋值后，返回给⽤户。

5.1 jinja2 的变量

from fastapi import FastAPI
import uvicorn
# 导包2
from fastapi.templating import Jinja2Templates

app = FastAPI()
# 模板位置
templates = Jinja2Templates(directory="templates")

# 导包1
from fastapi import Request


class game(object):
    def __init__(self, *args):
        self.game_list = args


@app.get("/index")
def index(request: Request):
    # 数据类型
    name = "蜡笔小新"
    age = 18
    books = [{"title": "金瓶梅", "price": 400},
             {"title": "聊斋", "price": 199},
             {"title": "剪灯新话", "price": 200},
             {"title": "国色天香", "price": 399},
             ]
    info = {"name": "蜡笔", "age": 24, "gender": "male"}
    pai = 3.1415926

    address = {"city": ["成都", "广州", "上海", "北京"], "movie": ["黑猫警长", "熊大熊二", "大头儿子"]}

    games = game("和平", "王者", "lol", "绝地求生")
    
    # 返回给模板的字典数据
    re_context = {
        # 必须有request参数
        "request": request,
        "user": name,
        "age": age,
        "books": books,
        "info": info,
        "pai": pai,
        "address": address,
        "games": games
    }

    return templates.TemplateResponse(
        "index.html",  # 模板文件
        context=re_context,  # 字典
    )


if __name__ == '__main__':
    uvicorn.run("run:app", port=8090, reload=True)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<p>用户名：{{ user|upper }}</p>
<p>年龄：{{ age }}</p>
<p>四大： {{ books }}</p>
<p><1>： {{ books.0 }}</p>
<p><2>： {{ books.1 }}</p>
<p><3>： {{ books.2 }}</p>
<p><4>： {{ books.3 }}</p>

<ul>
    <li> {{ books.0 }}</li>
    <li> {{ books.1 }}</li>
    <li> {{ books.2 }}</li>
    <li> {{ books.3 }}</li>
</ul>

<p>姓名：{{ info.name }}</p>
<p>年龄：{{ info.age }}</p>
<p>性别：{{ info.gender }}</p>

<p>pai : {{ pai|round(3) }}</p>
<p>pai : {{ pai|round(5) }}</p>

<p></p>

{% if age > 18 %}

    <ul>
        {% for addr in address.city %}
            <li>{{ addr }}</li>
        {% endfor %}
    </ul>

{% else %}

    <ul>
        {% for m in address.movie %}
            <li>{{ m }}</li>
        {% endfor %}
    </ul>

{% endif %}
<hr>
<ul>
    {% for book in books %}
        {% if book.price >= 200 %}
            <li>{{ book.title }}</li>
        {% endif %}
    {% endfor %}
</ul>

<ul>



            <li>{{ games.game_list}}</li>

</ul>
</body>
</html>

5.2 jinja2 的过滤器(内置函数)

过滤器名称	说明
capitialize	把值的⾸字母转换成⼤写，其他⼦母转换为⼩写
lower	把值转换成⼩写形式
title	把值中每个单词的⾸字母都转换成⼤写
trim	把值的⾸尾空格去掉
striptags	渲染之前把值中所有的HTML标签都删掉
join	拼接多个值为字符串
round	默认对数字进⾏四舍五⼊，也可以⽤参数进⾏控制
safe	渲染时值不转义

 
{{ 'abc'| captialize  }}  # Abc

{{ 'abc'| upper  }} # ABC

{{ 'hello world'| title  }} # Hello World

{{ "hello world"| replace('world','labi') | upper }} # HELLO LABI

{{ 18.18 | round | int }} # 18

5.3 jinja2 的条件和循环

{% if age > 18 %}
    <p>成年</p>
{% else %}
    <p>未成年</p>
{% endif %}




{% for book in books %}
    <p>{{ book }}</p>
{% endfor %}

---

6.ORM操作

Tortoise ORM文档
fastapi没有很好的数据库支持模块
pip install tortoise
pip install tortoise-orm

Tortoise ORM 目前支持以下数据库：

PostgreSQL >= 9.4（使用asyncpg）

SQLite（使用aiosqlite）

MySQL/MariaDB（使用aiomysql或使用asyncmy）

创建模型

安装
pip install tortoise
pip install tortoise-orm
pip install aiomysql

编写模型文件

from tortoise.models import Model
from tortoise import fields


class Clas(Model):
    name = fields.CharField(max_length=255, description='班级名称')


class Teacher(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, description='姓名')
    tno = fields.IntField(description='账号')
    pwd = fields.CharField(max_length=255, description='密码')


class Student(Model):
    id = fields.IntField(pk=True)
    sno = fields.IntField(description='学号')
    pwd = fields.CharField(max_length=255, description='密码')
    name = fields.CharField(max_length=255, description='姓名')
    # 一对多
    clas = fields.ForeignKeyField('models.Clas', related_name='students')
    # 多对多
    courses = fields.ManyToManyField('models.Course', related_name='students',description='学生选课表')


class Course(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, description='课程名')
    teacher = fields.ForeignKeyField('models.Teacher', related_name='courses', description='课程讲师')

aerich迁移工具

import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORM


app = FastAPI()

# 该方法会在fastapi启动时触发，内部通过传递进去的app对象，监听服务启动和终止事件
# 当检测到启动事件时，会初始化Tortoise对象，如果generate_schemas为True则还会进行数据库迁移
# 当检测到终止事件时，会关闭连接
register_tortoise(
    app,
    config=TORTOISE_ORM,
    # generate_schemas=True,  # 如果数据库为空，则自动生成对应表单，生产环境不要开
    # add_exception_handlers=True,  # 生产环境不要开，会泄露调试信息
)

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True,
               workers=1)

配置文件

TORTOISE_ORM = {
    'connections': {
        'default': {
            # 'engine': 'tortoise.backends.asyncpg',  PostgreSQL
            'engine': 'tortoise.backends.mysql',  # MySQL or Mariadb
            'credentials': {
                'host': '127.0.0.1',
                'port': '3306',
                'user': 'root',
                'password': 'yuan0316',
                'database': 'fastapi',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                "echo": True
            }
        },
    },
    'apps': {
        'models': {
             # 注意models的位置相对于当前
            'models': ['models', "aerich.models"],
            'default_connection': 'default',

        }
    },
    'use_tz': False,
    'timezone': 'Asia/Shanghai'
}

---

1. 初始化配置，只需要使用一次

aerich init -t settings.TORTOISE_ORM # TORTOISE_ORM配置的位置)

初始化完会在当前目录生成一个文件：pyproject.toml和一个文件夹：migrations

pyproject.toml：保存配置文件路径，低版本可能是aerich.ini

migrations：存放迁移文件

2. 初始化数据库，一般情况下只用一次

aerich init-db

1. 此时数据库中就有相应的表格
2. 如果TORTOISE_ORM配置文件中的models改了名，则执行这条命令时需要增加--app参数，来指定你修改的名字
   
   后续模型更新后的一些操作
   修改模型
   aerich migrate [–name] (标记修改操作) # aerich migrate –name add_column
   重新执行迁移，写入数据库
   aerich upgrade
   回到上一个版本
   aerich downgrade
   查看历史迁移记录
   aerich history

---

7.中间件

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

如果你使用了 yield 关键字依赖, 依赖中的退出代码将在执行中间件后执行.

如果有任何后台任务(稍后记录), 它们将在执行中间件后运行.

要创建中间件你可以在函数的顶部使用
@app.middleware("http").

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

request.

一个函数call_next，它将接收request，作为参数.

这个函数将 request 传递给相应的 路径操作.

然后它将返回由相应的路径操作生成的 response.

然后你可以在返回 response 前进一步修改它.

import uvicorn
from fastapi import FastAPI
# 导包
from fastapi import Request

from fastapi.responses import Response
import time

app = FastAPI()


@app.middleware("http")
async def m2(request: Request, call_next):
    # 请求代码块
    print("m2 request")
    response = await call_next(request)
    # 响应代码块
    response.headers["author"] = "yuan"
    print("m2 response")
    return response


@app.middleware("http")
async def m1(request: Request, call_next):
    # 请求代码块
    print("m1 request")
    # if request.client.host in ["127.0.0.1", ]:  # 黑名单
    #     return Response(content="visit forbidden")

    # if request.url.path in ["/user"]:
    #     return Response(content="visit forbidden")

    start = time.time()

    response = await call_next(request)
    # 响应代码块
    print("m1 response")
    end = time.time()
    response.headers["ProcessTimer"] = str(end - start)
    return response


@app.get("/user")
def get_user():
    time.sleep(3)
    print("get_user函数执行")
    return {
        "user": "current user"
    }


@app.get("/item/{item_id}")
def get_item(item_id: int):
    time.sleep(2)
    print("get_item函数执行")
    return {
        "item_id": item_id
    }


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

靠近路径函数的先执行

8.cors

import uvicorn
from fastapi import FastAPI
from fastapi import Request
# 导包
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 方式一
# @app.middleware("http")
# async def CORSMiddleware(request: Request, call_next):
#     response = await call_next(request)
#     response.headers["Access-Control-Allow-Origin"] = "*"
#     return response
origins = [
    "http://localhost:63342"
]
# 方式二
app.add_middleware(
    CORSMiddleware,
    allow_origins="*",  # *：代表所有客户端
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

@app.get("/user")
def get_user():
    print("user:xx", )
    return {
        "user": "xx"
    }


if __name__ == '__main__':
    uvicorn.run('run:app', host='127.0.0.1', port=8030, reload=True,
                 workers=1)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>

<p>hello world</p>


<script>
    $("p").click(function () {

        $.ajax({
            url: "http://127.0.0.1:8030/user",
            success: function (res) {
                console.log(res)
                console.log(res.user)
                $("p").html("hello " + res.user)


            }
        })

    })
</script>

</body>
</html>

总结

到最后你会发现 这个写接口真的好用,到此就结束了

作者：细精本精