FastAPI 介绍

FastAPI 与其它 Python-Web 框架的区别

在 FastAPI 之前，Python 的 Web 框架使用的是 django、flask、tornado 三种 Web 框架。

django 自带 admin，可快速构建，但是比较笨重。如果是 mvc 形式的开发，很多已经封装好了，的确蛮合适。但如果是 restful 风格设计，则 django 就显得有一些笨重了。
flask 快速构建，自由度高。因为它十分轻盈，插件即插即用，很适合用来做 restful 风格的设计
tornado Python Web 框架和异步网络库，它执行非阻塞 I/O , 没有对 REST API 的内置支持，但是用户可以手动实现。
FastAPI 快速构建，异步 IO，自带 Swagger 作为 API 文档，不用后续去内嵌 Swagger-Ui

我个人认为 FastAPI 是一个专门为 restful 风格设计，全面服务于 API 形式的 Web 后端框架。

FastAPI 官方定位

在 FastAPI 官方文档中，可以看到官方对 FastAPI 的定位：

快速：非常高的性能，向 NodeJS 和 go 看齐（感谢 Starlette 和 Pydantic)
快速编码：将功能开发速度提高约 200％至 300％。
错误更少：减少约 40％的人为错误（开发人员）。* (FastAPI 内置很多 Python 高版本的语法，比如类型注释，typing 库等等，因此被要求的 Python 版本为 3.6+)
简易：旨在易于使用和学习。减少阅读文档的时间。
功能完善: 自带 Swagger 作为 API 文档

Framework Benchmarks

https://www.techempower.com/benchmarks/#section=data-r19&hw=ph&test=fortune 1607394495_5fcee4bf504b3cb664ef0.png!small?1607394494984

1607394505_5fcee4c9a8be3dd6bda3f.png!small?1607394505452

1607394514_5fcee4d2a20d19d982e57.png!small?1607394514638

上图可以看出，在高并发下 4 个框架的排名情况。单纯从性能出发，Web 框架是排在第一的。在选用框架的时候，性能是一方面，我们还要看业务上的需求和使用场景，最适合的才是最好的。

下面简单介绍一下 FastAPI 的一些用法和特性.

启动FastAPI

# pip install fastapi

# pip install uvicorn

from fastapi import FastAPI

app = FastAPI()

@app.get("/")

def read_root():

    return {"Hello": "World"}

@app.get("/items/{item_id}")

def read_item(item_id: int, q: str = None):

    return {"item_id": item_id, "q": q}

# uvicorn main:app  # 启动

# uvicorn main:app --reload # 支持热更新

# uvicorn main:app --host 0.0.0.0 --port 8889 --reload # 自定义IP+端口

FastAPI 支持异步请求

from fastapi import FastAPI

app = FastAPI()

@app.get("/")

async def read_root():

    return {"Hello": "World"}

 

@app.get("/items/{item_id}")

async def read_item(item_id: int, q: str = None):

    return {"item_id": item_id, "q": q}

对 API 接口的支持性优异

设置根目录

# main.py

from fastapi import FastAPI

import users

app = FastAPI()

app.include_router(

    users.router,

    prefix="/fastapi/play/v1/users", # 路由前缀

    tags=['users'] # 路由接口类别

)

# routers/users.py

from fastapi import FastAPI,APIRouter

from datetime import datetime,timedelta

router = APIRouter()

@router.get("/get/users/")

async def get_users():

    return {

        "desc":"Return to user list"

    }

对路径参数进行限制

# 根据名字获取列表

@router.get("/get/user/{username}")

async def get_user_by_username(username :str):

    """

        - username: 用户名

    """

    return {

       "desc":"this username is "+ username

    }

对查询参数做限制

@router.get(“/friends/”)

# 设置为None的时候，默认不可以不填

async def get_friends_by_id(id :int=None):

    for item in test_json['friends']:

        if item['id'] == id:

            return item

        else:

            return {

                "desc": "no this id"

            }

# 多参数请求查询

from typing import List

@router.get("/items/")

async def read_items(q: List[str] = Query(["foo", "bar"])):

    query_items = {"q": q}

    return query_items

设置请求体

# 设置请求实体

from pydantic import BaseModel,Field

class requestModel(BaseModel):

    name :str

    age : int = Field(..., gt=0, description="The age must be greater than zero")

    desc: str
@router.post("/post/UserInfo/")

async def post_UserInfo(item: requestModel):

    return item

请求体嵌套

from pydantic import BaseModel,Field

class levelInfoModel(BaseModel):

    id:int = None

    info: str = None

 

class ClassInfo(BaseModel):

    id: int = None

    name: str = Field(..., max_length=20, min_length=10,

                      description="The necessary fields")

    desc: str = Field(None, max_length=30, min_length=10)

    levelInfo: List[levelInfoModel]

    class Config:

        schema_extra = {

            "example": {

                "id": 1,

                "name": "Foo",

                "desc": "A very nice Item",

                "levelInfo": [{

                    "id": 1,

                    "info": "一级"

                }]

            }

        }

 

@router.post("/info/")

async def get_classInfo(item:ClassInfo):

    return item

自定义响应码

@router.post("/items/", status_code=201)

async def create_item(name: str):

    return {"name": name}

from fastapi import FastAPI, status

@app.post("/items/", status_code=status.HTTP_201_CREATED)

async def create_item(name: str):

    return {"name": name}

依赖注入

from fastapi import Depends, FastAPI

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):

    return {"q": q, "skip": skip, "limit": limit}

@router.get("/items/")

async def read_items(commons: dict = Depends(common_parameters)):

    return commons

@router.get("/users/")

async def read_users(commons: dict = Depends(common_parameters)):

    return commons

FastAPI 框架支持多层嵌套依赖注入

登录demo

# 安装环境

mkdir fastapi-demo && cd fastapi-demo

virtualenv env

source env/bin/activate

# 下载项目

git clone https://github.com/hzjsea/BaseFastapi

cd BaseFastapi/

pip install -r requirements.txt

# 开启项目

uvicorn main:app --reload

# uvicorn main:app --host 0.0.0.0 --port 80 --reload

总结

FastAPI 的设计还是很符合 restful 的，在用到很多新技术的同时，也没有抛弃之前一些比较好用的内容，包括类型注释、依赖注入，Websocket，swaggerui 等等，以及其它的一些注释，比如 GraphQL。

数据库以及 orm 的选择

sqlalchemy 但是不支持异步，不过貌似可以扩展成异步。
tortoise-orm 类 django-orm 的异步 orm，不过正在起步过程中，有些功能还没有完成。

sqlalchemy实例

from typing import List

import databases

import sqlalchemy

from fastapi import FastAPI

from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app

DATABASE_URL = "sqlite:///./test.db"

# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(

    "notes",

    metadata,

    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

    sqlalchemy.Column("text", sqlalchemy.String),

    sqlalchemy.Column("completed", sqlalchemy.Boolean),

)

engine = sqlalchemy.create_engine(

    DATABASE_URL, connect_args={"check_same_thread": False}

)

metadata.create_all(engine)

 

 

class NoteIn(BaseModel):

    text: str

    completed: bool

 

 

class Note(BaseModel):

    id: int

    text: str

    completed: bool

 

 

app = FastAPI()

 

 

@app.on_event("startup")

async def startup():

    await database.connect()

 

 

@app.on_event("shutdown")

async def shutdown():

    await database.disconnect()

 

 

@app.get("/notes/", response_model=List[Note])

async def read_notes():

    query = notes.select()

    return await database.fetch_all(query)

 

 

@app.post("/notes/", response_model=Note)

async def create_note(note: NoteIn):

    query = notes.insert().values(text=note.text, completed=note.completed)

    last_record_id = await database.execute(query)

    return {**note.dict(), "id": last_record_id}

tortoise-orm实例

# main.py

from tortoise.contrib.fastapi import HTTPNotFoundError, register_tortois

# 创建的数据表

models = [

    "app.Users.models",

    "app.Face.models",

    "app.Roster.models",

    "app.Statistical.models",

    "app.pay.models"

]

 

register_tortoise(

    app,

    db_url="mysql://username:password@ip:port/yydb",

    modules={"models": models},

    generate_schemas=True,

    add_exception_handlers=True,

)

 

# models.py

from tortoise import fields,models

from tortoise.contrib.pydantic import pydantic_queryset_creator

from pydantic import BaseModel

class RosterGroupTable(models.Model):

    id = fields.IntField(pk=True)

    phone = fields.CharField(max_length=20,blank=True,null=True)

    name = fields.CharField(max_length=20)

 

    class Meta:

        db = "RosterGroupTable"

 

class RosterTabel(models.Model):

    id = fields.IntField(pk=True)

    phone = fields.CharField(max_length=20,blank=True,null=True)

    name = fields.CharField(max_length=20)

    group_id = fields.ForeignKeyField(model_name='models.RosterGroupTable',on_delete=fields.CASCADE,related_name="events",blank=True,null=True)

 

    class Meta:

        db = "RosterTabel"

 

RosterGroupTable_desc = pydantic_queryset_creator(RosterGroupTable)

RosterTabel_desc = pydantic_queryset_creator(RosterTabel)

 

 

 

# roster.py

@router.post("/roster_statistics/add")

async def roster_add_statics(*,item:RosterItem,token :str):

    res = await RosterTabel.filter(id=item['memberId']).first()

    if res:

        await StatisticalRosterTable.create(

            phone = current_user.Phone,

            date = item['date'],

            time = item['time'],

            data = item['data'],

            name_id_id = item['memberId'],

            temp_type = item['tm_type'],

            today_num = item['todayNum'],

            group_id_id = res.group_id_id,

            name = res.name

        )

    else:

        return rp_faildMessage(msg="名单不存在",code=-1)

    return rp_successMessage(msg="名单创建成功",code=0)

部署

dockerfile

#================================================================================

#      基于Python3.7的创建fastapi的dockerfile文件

#      fastapi + Python3.7 + guvicorn

#================================================================================

 

FROM Python:3.7

LABEL version="v1" \

    description="fastapi" \

    maintainer="hzjsea" \

    using="fastapi & Python3.7 office image & gunicorn"

WORKDIR /root/code

COPY  . .

RUN pip install -r requirements.txt

EXPOSE 8889

CMD ["gunicorn","-c","guvicorn.conf","main:app"]

单机本地部署

#!/usr/bin/bash

# 测试用脚本部署

 

# 开启的端口

port=80

host=0.0.0.0

 

# 删除旧的文件

rm -rf /root/hzj/fastapi_play

# 拷贝新的文件

cp -r /root/fastapi_play/ /root/hzj/

# 切换workdir

cd /root/hzj/fastapi_play/

# 开启python虚拟环境

source /root/hzj/pro_env_all/venv/bin/activate

# 安装依赖

pip install -r requirements.txt

# 关闭之前运行的进程  

lsof -i:$port | awk '{getline; print $2 }' | xargs -t -I {} kill -9 {}

# 开启新进程

nohup uvicorn --host $host  --port $port  main:app --reload  &

echo "success"

supervisor项目托管

[program:webserver]

directory=/root/hzj/fastapi_play

command=/root/hzj/pro_env_all/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8888

autostart = true

gunicorn部署asgi服务

# gunicorn.conf

并行工作进程数

workers = 4

# 指定每个工作者的线程数

threads = 2

# 监听内网端口5000

bind = '0.0.0.0:8889'

# 设置守护进程,将进程交给supervisor管理

daemon = 'false'

# 工作模式协程

worker_class = 'uvicorn.workers.UvicornWorker'

# 设置最大并发量

worker_connections = 2000

#运行

gunicorn -c gunicorn.conf