fastapi

学习 - FastAPI (tiangolo.com)

类型提示

first_name, last_name

first_name: str, last_name: str #用冒号

#有些容器数据结构可以包含其他的值，比如 dict、list、set 和 tuple。它们内部的值也会拥有自己的类型，可以使用typing库来声明
from typing import List

def process_items(items: List[str]):#嵌套类型
    for item in items:
        print(item)

#还可以把类声明为类型
class Person:
    def __init__(self, name: str):
        self.name = name


def get_person_name(one_person: Person):#one_person是Person类的
    return one_person.name

Pydantic模型

执行数据校验

你可以将数据的"结构"声明为具有属性的类。

每个属性都拥有类型。

接着你用一些值来创建这个类的实例，这些值会被校验，并被转换为适当的类型（在需要的情况下），返回一个包含所有数据的对象。

什么用？

• 编辑器支持

• 类型检查

• 定义参数要求

• 转换数据

• 校验数据

• 记录API

并发async await

@app.get('/')
async def read_results():
    results = await some_library()
    return results

'''你可以根据需要在路径操作函数中混合使用 def 和 async def，并使用最适合你的方式去定义每个函数。FastAPI 将为他们做正确的事情。'''

异步代码意味着告诉程序在某个点，代码必须停下来等待其他事情，这些事称为慢事件

等待慢事件期间可以做其他事，同时查看等待的事件是否有完成的。

这些慢事件多半是IO操作有关

现代版本的 Python 有一种非常直观的方式来定义异步代码。这使它看起来就像正常的"顺序"代码，并在适当的时候"等待"。

使用 async def，Python 就知道在该函数中，它将遇上 await，并且它可以"暂停"执行该函数，直至执行其他操作后回来。

@app.get('/burgers')
async def read_burgers():
    burgers = await get_burgers(2)
    return burgers

您可能已经注意到，await 只能在 async def 定义的函数内部使用。

但与此同时，必须"等待"通过 async def 定义的函数。因此，带 async def 的函数也只能在 async def 定义的函数内部调用。

那第一个async函数呢？

如果您使用 FastAPI，你不必担心这一点，因为"第一个"函数将是你的路径操作函数，FastAPI 将知道如何做正确的事情。

基础

一些有的没的

@app.get("/")表示当用户访问根路径（即网站的首页）时，会触发这个路由，并执行相应的处理函数。

uvicorn main:app --reload
# main：main.py 文件（一个 Python「模块」）。
# app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
# --reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

交互式api文档：http://127.0.0.1:8000/docs

from fastapi import FastAPI

app = FastAPI() #创造对象
@app.get("/helloworld")#get请求
async def root():
    return {"message": "Hello World"}


@app.get('/users/{user_id}')#localhost:8000/users/2  那么2就是user_id
async def get_user_id(user_id:int):
	return {'user_id':user_id}

路径参数

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/users/{user_id}")
async def get_id(user_id: int):
    return {"message": f'User ID: {user_id}'}

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

API的顺序影响

@app.get("/users/current")#固定格式，小范围，先
async def get_current_user():
	return{"user_id":"current_id"}

@app.get("/users/{user_id}")#匹配格式，大范围，后
async def get_user_id(user_id:int):
	return {"user_id":user_id}

#匹配顺序很重要

配合swaggerui

from enum import Enum
from fastapi import FastAPI
import uvicorn

app = FastAPI()

class Gender(str, Enum):#自定义一个类，枚举Enum
    male = "male"
    female = "female"

@app.get("/users/{user_id}")
async def get_id(user_id: int):
    return {"message": f'User ID: {user_id}'}

@app.get("/students/{gender}")
async def get_gender(gender: Gender):#gender是Gender类
    return {"gender": f'{gender.value}'}


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

查询参数

http://localhost:8000/users?page_index=1&page_size=10

from fastapi import FastAPI
import uvicorn

app = FastAPI()
@app.get("/users")#查询参数不用在url里写了
async def ger_user(page_index:int, page_size:int):
    return {"Page Info": f'Page Index: {page_index}, Page Size: {page_size}'}


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

可选查询参数

@app.get("/users")
async def ger_user(page_index:int, page_size:Optional[int]=10):#默认10，可选输入
    return {"Page Info": f'Page Index: {page_index}, Page Size: {page_size}'}

混合参数

@app.get("/users/{user_id}/friends")#查询参数不用在url里写了
async def ger_user(user_id:int,page_index:int, page_size:Optional[int]=10):#默认10，可选输入
    return {"Info": f'user_id: {user_id}, Page Index: {page_index}, Page Size: {page_size}'}

请求体

类型可以是post，put，delete，patch

from pydantic import BaseModel

class UserModel(BaseModel):
    username:str
    description:Optional[str]=None

from enum import Enum
from typing import Optional
from fastapi import FastAPI
import uvicorn
from pydantic import BaseModel


class Gender(str, Enum):#枚举，只能这两种
    male = "male"
    female = "female"


class UserModel(BaseModel):
    username:str
    description:Optional[str]='Default'#可选项,并且设置了一个默认值
    gender:Gender

app = FastAPI()

@app.post("/users")#提交
async def create_user(user_model:UserModel):#自动翻译到对象里
    print(user_model.username)#说明已经发送成功，打印在终端
    user_dict=user_model.model_dump()#把模型里的数据转成字典
    return user_dict

@app.put("/users/{user_id}")#修改
async def update_user(user_id:int,user_model:UserModel):#自动翻译到对象里
    print(user_model.username)#说明已经发送成功，打印在终端
    user_dict=user_model.model_dump()#把模型里的数据转成字典
    user_dict.update({"user_id":user_id})
    return user_dict#返回到客户端

#请求体有多个模型参数
class user(BaseModel):
    username:str
    description:Optional[str] = 'default'

class item(BaseModel):
    name:str
    price:float
    is_offer:bool = None

'''
{
  "user": {
    "username": "string",
    "description": "default"
  },
  "item": {
    "name": "string",
    "price": 0,
    "is_offer": true
  }
}
'''

@app.put('/carts/{cart_id}')
async def cupdate_cart(cart_id : int, user:user, item:item):
    print(user.username)
    print(item.name)
    result_dict={
        'cart_id':cart_id,
        'user_name':user.username,
        'item_name':item.name
    }

    return result_dict
###
###
###
@app.put('/carts/{cart_id}')
async def cupdate_cart(cart_id : int, user:user, item:item,count:int=Body(...,ge=2)):#这样count就识别为请求体里的，不这么写就会识别为查询参数,并且还有验证
    print(user.username)
    print(item.name)
    result_dict={
        'cart_id':cart_id,
        'user_name':user.username,
        'item_name':item.name
    }
###
###
###
class user(BaseModel):
    username:str = Field(...,min_length=2,max_length=20)#请求体模型内的验证
    description:Optional[str] = 'default'
###
###
###

参数验证

路径参数：fastapi.Path

查询参数：fastapi.Query

@app.get("/users/{user_id}")#查询参数不用在url里写了
async def ger_user(user_id:int = Path(...,title='user id',ge=1,le=1000)): #Path验证路径参数，...前在前面表示必须有user_id，ge大于等于，le小于等于，还有min_length,max_length,lt小于,gt大于
    return {"Info": f'user_id: {user_id}'}


@app.get("/items/{item_name}")#查询参数不用在url里写了
async def ger_item(item_name:str = Path(...,title='item name',regex='^[a-z]-[\d]+$')): #Path验证路径参数，...前在前面表示必须有item_name，regex表示正则表达式
    return {"Info": f'item_name: {item_name}'}


Query(None):查询参数是可选项（本来），其实是定义了默认值，只不过是None而已，也可以是1，2，3...即不写就用默认的
Query(...):这下就是必选的了


@app.get("/items")#查询参数不用在url里写了
async def ger_item(page_index : int = Query(1,alias='page-index',title='page_index',ge=1,le=1000)): #指定用别名验证，url中必须是page-index
    return {"Info": f'page_index: {page_index}'}

###
###
###
@app.put('/carts/{cart_id}')
async def cupdate_cart(cart_id : int, user:user, item:item,count:int=Body(...,ge=2)):#这样count就识别为请求体里的，不这么写就会识别为查询参数,并且还有验证
    print(user.username)
    print(item.name)
    result_dict={
        'cart_id':cart_id,
        'user_name':user.username,
        'item_name':item.name
    }
###
###
###
class user(BaseModel):
    username:str = Field(...,min_length=2,max_length=20)#请求体模型内的验证
    description:Optional[str] = 'default'
###
###
###

示例数据

显示在文档里的例子

class user(BaseModel):
    username:str = Field(...,min_length=2,max_length=20,examples=['Jack'])#[]是为了兼容，有一些是List，有一些是一个值
    description:Optional[str] = Field(None, max_length=100,examples=['This is Jack'])

###
###
###
@app.put('/carts/{cart_id}')
async def cupdate_cart(*,cart_id : int, user:user=Body(...,examples=[{'username':'Smith'}]), item:item,count:int=Body(...,ge=2)):#不加*，就会出现错误，第二个给了默认值，而前后都没有，如果加了*，那么后面的参数就都变成关键字参数，而不是顺序参数，也就是传参必须写明了参数名
    print(user.username)
    print(item.name)
    result_dict={
        'cart_id':cart_id,
        'user_name':user.username,
        'item_name':item.name
    }

cookie与header

@app.put('/carts')
async def update_cart(*,response:Response,favorite_schema : Optional[str] = Cookie(None,alias='favorite-schema'),api_token : Union[str, None] = Header(None,alias='api-token')):
    result_dict={
        'favorite_schema':favorite_schema,
        'api_token':api_token
    }
    #可以专门写个api来set cookie
    response.set_cookie(key='favorite-schema',value='dark')#返回response后设定cookie，第一次不发favorite-schema，第二次自己设定

    return result_dict

响应模型

class UserOut(BaseModel):
    id:int
    username:str
    description:Optional[str]='default'

users={
    'x':{'id':0},
    'a':{'id':1,'username':'a'},
    'b':{'id':2,'username':'b','password':'123456'},
    'c':{'id':3,'username':'c','password':'123456','description':'this is c'},
    'd':{'id':4,'username':'d','password':'123456','description':'this is d'},
    'e':{'id':5,'username':'e','password':'123456','description':'this is e','fullname':'Jack Smith'}
    }

@app.get('/users/{username}',response_model=UserOut)#输出结果用这个模型去规范
async def get_user(username: str):
    return users.get(username,{})#{}表示找不到就返回空,返回由username匹配的一整项

###
@app.get('/users/{username}',response_model=UserOut,response_model_include={'id'})#输出结果用这个模型去规范,但只输出id
###

@app.get('/users',response_model=List[UserOut])#输出列表，用模型规范
async def get_users():
     return users.values()

状态码与异常处理

users={
    # 'x':{'id':0},
    'a':{'id':1,'username':'a'},
    'b':{'id':2,'username':'b','password':'123456'},
    'c':{'id':3,'username':'c','password':'123456','description':'this is c'},
    'd':{'id':4,'username':'d','password':'123456','description':'this is d'},
    'e':{'id':5,'username':'e','password':'123456','description':'this is e','fullname':'Jack Smith'}
    }

@app.get('/users/{username}',status_code=200, response_model=UserOut)#成功就200
async def get_users(username:str=Path(...,min_length=1)):
    user=users.get(username,None)
    if user:
        return user#返回的是个字典或者说jSON
    
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail='User not found')#失败就404


###自定义异常
class UserNotFoundException(Exception):
    def __init__(self,username : str):
        self.username = username

@app.get('/users/{username}',status_code=200, response_model=UserOut)#成功就200
async def get_users(username:str=Path(...,min_length=1)):
    user=users.get(username,None)
    if user:
        return user#返回的是个字典或者说JSON
    
    raise UserNotFoundException(username)#抛出异常

@app.exception_handler(UserNotFoundException)#一旦出现这个异常，就会调用这个函数
async def user_not_found_exception_handler(request:Request,exc:UserNotFoundException):#接受请求，接受异常对象
    return JSONResponse(status_code=404,content={
        'error_code':404,
        'message':f'User {exc.username} not found',
        'info':'ohno!'
    })

###处理多种异常
class ErrorMessage(BaseModel):
    error_code : int
    message : str

@app.post('/users',status_code=201,response_model=UserOut,responses={
    400:{"model":ErrorMessage},
    401:{"model":ErrorMessage} 
})#成功就201，那其他呢？这里就可以给其他情况了#胡乱写的,401应该是另一个model
async def create_user(user:UserIn):
    if users.get(user.username,None):#找不到就返回空，找到就报下面的错，暂时没法直接用ErrorMessage去规范输出，所以又去新建了error_message
        error_message=ErrorMessage(error_code=400,message=f'User {user.username} already exists')
        return JSONResponse(status_code=400,content=error_message.model_dump())#是个对象，我就把内容倒出来，倒出来个JSON格式
    user_dict=user.model_dump()
    user_dict.update({'id':10})
    return user_dict#这里的返回会直接被UserOut这个模型去规范

依赖注入

函数作为依赖项

def pageinfo_params(page_index : Optional[int]=1,page_size : Optional[int]=10):
    return {'page_index':page_index,'page_size':page_size}


@app.get('/items')
async def get_items(page_info : dict= Depends(pageinfo_params)):#原本写在api里的放到一个函数里去了，这里就注入依赖Depends
    
    return {'page_index':page_info.get('page_index'),'page_size':page_info.get('page_size')}

@app.get('/users')
async def get_users(page_info : dict= Depends(pageinfo_params)):#代码可以复用了，只需要depends一下
    
    return {'page_index':page_info.get('page_index'),'page_size':page_info.get('page_size')}
    

类作为依赖项

class PageInfo:
    def __init__(self,page_index : Optional[int]=1,page_size : Optional[int]=10):
        self.page_index = page_index
        self.page_size = page_size

@app.get('/users')
async def get_users(page_info : PageInfo = Depends(PageInfo)):#依赖一个类，PageInfo类新建了一个对象，这个对象的类型是PageInfo
    
    return {'page_index':page_info.page_index,'page_size':page_info.page_size}

@app.get('/goods')
async def get_users(page_info : PageInfo = Depends()):#后面的可以不用写，自动用前面那个
    
    return {'page_index':page_info.page_index,'page_size':page_info.page_size}


###
###
###路径注解中使用dependencies
async def verify_auth(api_token : Optional[str]=Header(None,alias='api-token')):#验证身份
    if not api_token:
        raise HTTPException(status_code=400,detail='unauthorized')

@app.get('/users',dependencies=[Depends(verify_auth)])#依赖于verify_auth，这个函数不能在下面的函数里使用，这是一个特点，[]表示可以有多个depends，逗号隔开
async def get_users(page_info : PageInfo = Depends(PageInfo)):
    
    return {'page_index':page_info.page_index,'page_size':page_info.page_size}

###
###
###还可以在建立FastApi对象时引用依赖，全局dependencies
async def set_charset():#在最前面
    print('set utf-8')

app = FastAPI(dependencies=[Depends(set_charset)])

身份认证

数据库建立和查询

from typing import List
import uvicorn
from fastapi import Depends, FastAPI,Path,Body,Query,HTTPException
from pydantic import BaseModel,Field
from sqlalchemy import Column, Integer, asc, create_engine,MetaData,Table,select,String
from sqlalchemy.orm import sessionmaker,Mapped,DeclarativeBase,mapped_column

class Base(DeclarativeBase):
    pass

engine = create_engine('mysql://root:root@localhost:3306/testdb')

class StudentEntity(Base):
    __tablename__ = 'student'
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50),unique=True,nullable=False)
    age: Mapped[int] = mapped_column(Integer)
    gender: Mapped[str] = mapped_column(String(10),nullable=False)


Base.metadata.create_all(engine)#根据映射创建表
Session = sessionmaker(bind=engine)
app = FastAPI()

class StudentBase(BaseModel):
    name : str
    gender : str

class StudentCreate(StudentBase):
    ...

class StudentOut(StudentBase):
    id : int

#用Depend创建数据库Session
def get_db_session():
    db_session = Session()
    try:
        yield db_session
    finally:
        db_session.close()

@app.get('/students',response_model=List[StudentOut])
async def get_students(db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).order_by(asc(StudentEntity.id))
    return db_session.execute(query).scalars().all()

@app.post('/students',response_model=StudentOut)#建表
async def create_student(student : StudentCreate,db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).where(StudentEntity.name == student.name)#若姓名已存在
    records = db_session.execute(query).scalars().all()#执行查询操作
    if records:
        raise HTTPException(status_code=400,detail='Student {student.name} already exists')
    
    student_entity = StudentEntity(name=student.name,gender=student.gender)#使用传入值建表
    db_session.add(student_entity)#执行
    db_session.commit()#提交
    db_session.refresh(student_entity)
    return student_entity

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

数据库修改和删除

from typing import List
import uvicorn
from fastapi import Depends, FastAPI,Path,Body,Query,HTTPException
from pydantic import BaseModel,Field
from sqlalchemy import Column, Integer, asc, create_engine,MetaData,Table,select,String
from sqlalchemy.orm import sessionmaker,Mapped,DeclarativeBase,mapped_column


def set_attrs(obj,data : dict):
    if data:
        for key,value in data.items():#取出字典中所有数据
            setattr(obj,key,value)

class Base(DeclarativeBase):
    pass

engine = create_engine('mysql://root:root@localhost:3306/testdb')

class StudentEntity(Base):
    __tablename__ = 'student'
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50),unique=True,nullable=False)
    age: Mapped[int] = mapped_column(Integer)
    gender: Mapped[str] = mapped_column(String(10),nullable=False)


Base.metadata.create_all(engine)#根据映射创建表
Session = sessionmaker(bind=engine)
app = FastAPI()

class StudentBase(BaseModel):
    name : str
    gender : str

class StudentCreate(StudentBase):
    ...

class StudentUpdate(StudentBase):
    ...

class StudentOut(StudentBase):
    id : int

#用Depend创建数据库Session
def get_db_session():
    db_session = Session()
    try:
        yield db_session
    finally:
        db_session.close()

@app.get('/students',response_model=List[StudentOut])
async def get_students(db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).order_by(asc(StudentEntity.id))
    return db_session.execute(query).scalars().all()

@app.post('/students',response_model=StudentOut)#建表
async def create_student(student : StudentCreate,db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).where(StudentEntity.name == student.name)#若姓名已存在
    records = db_session.execute(query).scalars().all()#执行查询操作
    if records:
        raise HTTPException(status_code=400,detail='Student {student.name} already exists')
    
    student_entity = StudentEntity(name=student.name,gender=student.gender)#使用传入值建表
    db_session.add(student_entity)#执行
    db_session.commit()#提交
    db_session.refresh(student_entity)
    return student_entity

@app.put('/students/{student_id}',response_model=StudentOut)
async def update_student(*,student_id : int = Path(...),
                         student : StudentUpdate,
                         db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).where(StudentEntity.id == student.id)#找有没有
    exist_student = db_session.execute(query).scalar()#最多一条记录，就不scalars()了
    if not exist_student:
        raise HTTPException(status_code=404,detail=f'Student {student_id} not found')
    
    # exist_student.name = student.name
    # exist_student.gender = student.gender#过于繁琐

    set_attrs(exist_student,student.model_dump())#用model_dump()把student转成dict，再set_attrs()

    db_session.commit()
    return exist_student


@app.delete('/students/{student_id}')
async def delete_student(*,student_id : int = Path(...),
                         db_session : Session = Depends(get_db_session)): # type: ignore
    query = select(StudentEntity).where(StudentEntity.id == student_id)
    exist_student = db_session.execute(query).scalar()
    if not exist_student:
        raise HTTPException(status_code=404,detail=f'Student {student_id} not found')
    
    db_session.delete(exist_student)

    return {'detail': f'Student {student_id} deleted'}


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