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

# pip install fastapi# pip install uvicornfrom fastapi import FastAPIapp = 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+端口

from fastapi import FastAPIapp = 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}

# main.pyfrom fastapi import FastAPIimport usersapp = FastAPI()app.include_router(    users.router,    prefix="/fastapi/play/v1/users", # 路由前缀    tags=['users'] # 路由接口类别)# routers/users.pyfrom fastapi import FastAPI,APIRouterfrom datetime import datetime,timedeltarouter = 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,Fieldclass 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,Fieldclass 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

# 安装环境mkdir fastapi-demo && cd fastapi-demovirtualenv envsource 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

from typing import Listimport databasesimport sqlalchemyfrom fastapi import FastAPIfrom pydantic import BaseModel# SQLAlchemy specific code, as with any other appDATABASE_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}

# main.pyfrom 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.pyfrom tortoise import fields,modelsfrom tortoise.contrib.pydantic import pydantic_queryset_creatorfrom pydantic import BaseModelclass 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)

#================================================================================#      基于Python3.7的创建fastapi的dockerfile文件#      fastapi + Python3.7 + guvicorn#================================================================================
FROM Python:3.7LABEL version="v1" \    description="fastapi" \    maintainer="hzjsea" \    using="fastapi & Python3.7 office image & gunicorn"WORKDIR /root/codeCOPY  . .RUN pip install -r requirements.txtEXPOSE 8889CMD ["gunicorn","-c","guvicorn.conf","main:app"]