【Fastapi框架】Fastapi的使用和进阶

原文作者：我辈李想
版权声明：文章原创，转载时请务必加上原文超链接、作者信息和本声明。

前言

一、项目启动

1.安装

pip install fastapi  uvicorn

2.示例

新建main.py文件，

from typing import Union

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: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

3.启动

方式一：命令启动

uvicorn main:app --reload

方式二:在mian.py中加启动代码

# mian.py

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )

4.路由

http://127.0.0.1:8000/
http://127.0.0.1:8000/items/1
http://127.0.0.1:8000/items/1?q=2
http://127.0.0.1:5678/docs#/
http://127.0.0.1:5678/redoc

二、进阶

1.请求数据

1.get请求:路径参数+查询参数

路由：http://127.0.0.1:8000/items/1?q=2，其中1是路径参数，q=2是查询参数。

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

2.post/put/delete请求：路径参数+请求体

路由：http://127.0.0.1:8000/items/1，其中1是路径参数，item是请求体。pydantic 可以验证输入和输出。

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: Union[bool, None] = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

3.嵌套数据

# 1.定义基础模型
class Image(BaseModel):
    url: str

# 2.赋给其他类属性
class Item(BaseModel):
    image: Union[Image, None] = None  # 单个
    images: Union[List[Image], None] = None  # 多个

4.文件上传

from fastapi import FastAPI, status, File, Form, UploadFile

app = FastAPI()

# 方法一：上传的文件会存在内存中，适合小型文件 
@app.post("/api/uploadFile/action")
async def create_file(file: Annotated[bytes, File()]):  
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 方法二：UploadFile
@app.post("/api/uploadFile/action")
async def create_upload_file(file: UploadFile):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 上传多个文件
@app.post("/api/uploadFile/action")
async def create_upload_files(files: list[UploadFile]):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 用 File() 添加额外设定
@app.post("/api/uploadFile/action")
async def create_upload_files(
    files: Annotated[
        list[UploadFile], File(description="Multiple files as UploadFile")
    ],
):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 将file写入dirs目录文件
def writeBytes(dirs,file):
    bytesFile=file.file.read()
    filename=file.filename
    if not os.path.exists(dirs):
        os.makedirs(dirs)
    with open(dirs+'/'+ filename, "wb") as f:
        f.write(bytesFile)

5.数据校验

使用fastapi.params库中的函数，FastAPI会自动验证请求参数。Query，Bath 等。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query(None, min_length=3, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

6.request参数

@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
    args = await request.form()  # form数据
    args2 = await request.json()   # json数据
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}

7.文档注释

@router.post('/roll-plan/kpi/', summary='获取输入数据', description='', tags=['算法'])
async def get_roll_plan(times: Annotated[RollPlanTimeItem, Body(examples=[{"start_time": "2023-12-12 00:00:00",
                                                                           "end_time": "2023-12-13 00:00:00"}],
                                                                )]
                        ):

8.请求体对象转json

from datetime import datetime
from typing import Union

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    # item.json()
    # item.dict()
    fake_db[id] = json_compatible_item_data

9.sqlalchemy查询数据转json

roll_plans = session.query(rollPlan.AppRollPlanEntity).filter(
	rollPlan.AppRollPlanEntity.StartTime >= time_range[0],
	rollPlan.AppRollPlanEntity.StartTime <= time_range[1]).all()
	
# 方式一
from sqlalchemy.ext.serializer import loads, dumps
def to_dict(row):
    row_dict = row.__dict__
    if '_sa_instance_state' in row_dict:
        del row_dict['_sa_instance_state']
    return row_dict

serialized_data = [to_dict(result) for result in roll_plans]  # 将每个结果转换成字典形式
json_data = dumps(serialized_data)  # 将字典序列化为JSON字符串
# 反序列化
deserialized_data = loads(serialized_data)
user = User(**deserialized_data)

# 方式二
import json
import decimal

class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            print("MyEncoder-datetime.datetime")
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        if isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        if isinstance(obj, int):
            return int(obj)
        elif isinstance(obj, float):
            return float(obj)
        elif isinstance(obj, decimal.Decimal):
            return float(obj)
        # elif isinstance(obj, decimal):
        #    return decimal(obj)
        else:
            return super(MyEncoder, self).default(obj)
            
def to_dict(row):
    row_dict = row.__dict__
    if '_sa_instance_state' in row_dict:
        del row_dict['_sa_instance_state']
    return row_dict

serialized_data = [to_dict(result) for result in roll_plans]  # 将每个结果转换成字典形式
json_data = json.dumps(serialized_data,cls=MyEncoder, ensure_ascii=False)  # 将字典序列化为JSON字符串

2.静态文件加载

from fastapi import FastAPI
from starlette.staticfiles import StaticFiles

app = FastAPI()

# 静态文件配置使用mount挂载 第一个参数为应用的前缀，第二个为路径，第三个为别名
app.mount('/static', StaticFiles(directory='static'), name='static')

这里需要再main.py的同级目录创建static文件，在html文件中需要使用url_for引入静态文件。

<link rel="stylesheet" href="{{ url_for('static',path='demo.css') }}">

3.路由管理

1.在maipn.py的同级创建user文件夹,新建userRouter.py文件

from typing import Union

from fastapi import FastAPI, APIRouter, Request

router = APIRouter(prefix='/bp', tags=['bp'])

@router.get('/')
def read_root():
    return {"Hello": "World"}

@router.get('/items/{item_id}')
async def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}


# @router.get('/test/{apiname}')
# @router.post("/test/{apiname}")
@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
	args = await request.form()
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}

2.修改mapin.py内容

from fastapi import FastAPI
from user import userRouter 

app = FastAPI()
app.include_router(userRouter.router)

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )

4.跨域配置

# 跨域配置
origins = [
    "http://localhost:3000",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

5.自定义中间件

1.已有中间件

HTTPSRedirectMiddleware是fasapi自带的中间件

app.add_middleware(HTTPSRedirectMiddleware)
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["tigeriaf.com", "*.tigeriaf.com"]
)

2.自定义中间件

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

6.使用jwt中间件

在 FastAPI 中使用 JWT 中间件，可以通过自定义 FastAPI 的中间件来实现。以下是在 FastAPI 中使用 JWT 中间件的步骤。

1. 安装 PyJWT。

pip install PyJWT

2. 导入 PyJWT 库和 FastAPI 组件。

import jwt
from fastapi import FastAPI, HTTPException
from fastapi.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED

3. 定义一些全局变量，例如：

SECRET_KEY = "mysecretkey"  # 生产环境中应使用更复杂的密钥
ALGORITHM = "HS256"  # 哈希算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 访问令牌的过期时间，单位为分钟

4. 定义 JWT 认证的中间件函数。

async def auth_middleware(request: Request, call_next):
    try:
        token: str = request.headers["Authorization"].split()[1]
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
        request.state.user = username
    except (KeyError, IndexError, jwt.JWTError):
        raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
    response = await call_next(request)
    return response

auth_middleware 函数获取请求头中的访问令牌，并解码 JWT。如果 JWT 解码成功，则将用户名添加到请求对象的 state 属性中。如果 JWT 解码失败，则抛出 HTTPException 异常。

5. 在 FastAPI 中添加中间件。

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
    Middleware(auth_middleware)
]

app = FastAPI(
    title="My API",
    openapi_url="/api/v1/openapi.json",
    docs_url="/api/v1/docs",
    redoc_url=None,
    middleware=middleware
)

在 FastAPI 的构造函数中，使用 Middleware 类添加 CORSMiddleware 和 auth_middleware 中间件。auth_middleware 中间件将在每个路由处理函数执行前进行验证。因此，只需将路由的处理函数包装在 async def 函数中即可。

@app.get("/protected")
async def protected_route(request: Request):
    current_user = request.state.user
    return {"message": "Hello, {}".format(current_user)}

在以上示例中，使用 Request 类作为处理函数的一个参数，然后从请求的 state 属性中获取用户。

以上是使用 JWT 中间件的步骤，你可以根据项目的需求对上述代码进行进一步修改和优化。文章来源地址https://www.toymoban.com/news/detail-759267.html

到了这里，关于【Fastapi框架】Fastapi的使用和进阶的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！