2023面试高手

这个屌丝很懒，什么也没留下！

热门标签

python3异步框架_三分钟了解 Python3 的异步 Web 框架 FastAPI

作者：2023面试高手 | 2024-03-06 13:04:09

踩

python @router.post

快速编码，功能完善。从启动到部署，实例详解异步 py3 框架选择 FastAPI 的原因。

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

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

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

启动FastAPI

1 # pip install fastapi

2 # pip install uvicorn

3 from fastapi import FastAPI

4 app = FastAPI()

5 @app.get("/")

6 def read_root():

7 return {"Hello": "World"}

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

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

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

11 # uvicorn main:app # 启动

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

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

FastAPI 支持异步请求

1 from fastapi import FastAPI

2 app = FastAPI()

3 @app.get("/")

4 async def read_root():

5 return {"Hello": "World"}

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

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

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

对 API 接口的支持性优异

设置根目录

1 # main.py

2 from fastapi import FastAPI

3 import users

4 app = FastAPI()

5 app.include_router(

6 users.router,

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

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

9 )

10 # routers/users.py

11 from fastapi import FastAPI,APIRouter

12 from datetime import datetime,timedelta

13 router = APIRouter()

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

15 async def get_users():

16 return {

17 "desc":"Return to user list"

18 }

对路径参数进行限制

1 # 根据名字获取列表

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

3 async def get_user_by_username(username :str):

4 """

5 - username: 用户名

6 """

7 return {

8 "desc":"this username is "+ username

9 }

对查询参数做限制

1 @router.get("/friends/")

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

4 async def get_friends_by_id(id :int=None):

5 for item in test_json['friends']:

6 if item['id'] == id:

7 return item

8 else:

9 return {

10 "desc": "no this id"

11 }

12 # 多参数请求查询

13 from typing import List

14 @router.get("/items/")

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

16 query_items = {"q": q}

17 return query_items

设置请求体

1 # 设置请求实体

2 from pydantic import BaseModel,Field

3 class requestModel(BaseModel):

4 name :str

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

6 desc: str

9 @router.post("/post/UserInfo/")

10 async def post_UserInfo(item: requestModel):

11 return item

请求体嵌套

1 from pydantic import BaseModel,Field

2 class levelInfoModel(BaseModel):

3 id:int = None

4 info: str = None

6 class ClassInfo(BaseModel):

7 id: int = None

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

9 description="The necessary fields")

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

11 levelInfo: List[levelInfoModel]

13 class Config:

14 schema_extra = {

15 "example": {

16 "id": 1,

17 "name": "Foo",

18 "desc": "A very nice Item",

19 "levelInfo": [{

20 "id": 1,

21 "info": "一级"

22 }]

23 }

24 }

26 @router.post("/info/")

27 async def get_classInfo(item:ClassInfo):

28 return item

自定义响应码

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

2 async def create_item(name: str):

3 return {"name": name}

5 from fastapi import FastAPI, status

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

9 async def create_item(name: str):

10 return {"name": name}

依赖注入

1 from fastapi import Depends, FastAPI

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

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

6 @router.get("/items/")

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

8 return commons

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

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

12 return commons

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

登录demo

1 # 安装环境

2 mkdir fastapi-demo && cd fastapi-demo

3 virtualenv env

4 source env/bin/activate

6 # 下载项目

7 git clone https://github.com/hzjsea/Bas...

8 cd BaseFastapi/

9 pip install -r requirements.txt

10 # 开启项目

11 uvicorn main:app --reload

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

总结

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

数据库以及 orm 的选择

sqlalchemy 但是不支持异步，不过貌似可以扩展成异步。

tortoise-orm 类 django-orm 的异步 orm，不过正在起步过程中，有些功能还没有完成。

sqlalchemy实例

1 from typing import List

2 import databases

3 import sqlalchemy

4 from fastapi import FastAPI

5 from pydantic import BaseModel

6 # SQLAlchemy specific code, as with any other app

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

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

9 database = databases.Database(DATABASE_URL)

10 metadata = sqlalchemy.MetaData()

11 notes = sqlalchemy.Table(

12 "notes",

13 metadata,

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

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

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

17 )

18 engine = sqlalchemy.create_engine(

19 DATABASE_URL, connect_args={"check_same_thread": False}

20 )

21 metadata.create_all(engine)

24 class NoteIn(BaseModel):

25 text: str

26 completed: bool

29 class Note(BaseModel):

30 id: int

31 text: str

32 completed: bool

35 app = FastAPI()

38 @app.on_event("startup")

39 async def startup():

40 await database.connect()

43 @app.on_event("shutdown")

44 async def shutdown():

45 await database.disconnect()

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

49 async def read_notes():

50 query = notes.select()

51 return await database.fetch_all(query)

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

55 async def create_note(note: NoteIn):

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

57 last_record_id = await database.execute(query)

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

tortoise-orm实例

1 # main.py

2 from tortoise.contrib.fastapi import HTTPNotFoundError, register_tortois

3 # 创建的数据表

4 models = [

5 "app.Users.models",

6 "app.Face.models",

7 "app.Roster.models",

8 "app.Statistical.models",

9 "app.pay.models"

10 ]

12 register_tortoise(

13 app,

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

15 modules={"models": models},

16 generate_schemas=True,

17 add_exception_handlers=True,

18 )

20 # models.py

21 from tortoise import fields,models

22 from tortoise.contrib.pydantic import pydantic_queryset_creator

23 from pydantic import BaseModel

24 class RosterGroupTable(models.Model):

25 id = fields.IntField(pk=True)

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

27 name = fields.CharField(max_length=20)

29 class Meta:

30 db = "RosterGroupTable"

32 class RosterTabel(models.Model):

33 id = fields.IntField(pk=True)

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

35 name = fields.CharField(max_length=20)

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

38 class Meta:

39 db = "RosterTabel"

41 RosterGroupTable_desc = pydantic_queryset_creator(RosterGroupTable)

42 RosterTabel_desc = pydantic_queryset_creator(RosterTabel)

46 # roster.py

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

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

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

50 if res:

51 await StatisticalRosterTable.create(

52 phone = current_user.Phone,

53 date = item['date'],

54 time = item['time'],

55 data = item['data'],

56 name_id_id = item['memberId'],

57 temp_type = item['tm_type'],

58 today_num = item['todayNum'],

59 group_id_id = res.group_id_id,

60 name = res.name

61 )

62 else:

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

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

部署

dockerfile

1 #================================================================================

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

3 # fastapi + Python3.7 + guvicorn

4 #================================================================================

6 FROM Python:3.7

7 LABEL version="v1" \

8 description="fastapi" \

9 maintainer="hzjsea" \

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

11 WORKDIR /root/code

12 COPY . .

13 RUN pip install -r requirements.txt

14 EXPOSE 8889

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

supervisor项目托管

1 [program:webserver]

2 directory=/root/hzj/fastapi_play

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

4 autostart = true

部署完整示例

文档及项目地址：