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 Benchmarkshttps://www.techempower.com/benchmarks/#section=data-r19&hw=ph&test=fortune上图可以看出，在高并发下 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: [email protected]("/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 commonsFastAPI 框架支持多层嵌套依赖注入登录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 = truegunicorn部署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部署完整示例FastAPI官方提供了一个前后端分离项目完整示例https://github.com/tiangolo/full-stack-fastapi-postgresql文档及项目地址：Documentation: https://fastapi.tiangolo.com