FastAPI 學習筆記

最近在公司中需要寫接口，選取了fastapi這個框架，一個原因是FastAPI 是主流框架，同時FastAPI 有著高性能，支持異步和高并發。

FastAPI 安裝

直接用下面兩行命令進行安裝

pip3 install fastapi
pip install uvicorn

FastAPI 例子

下面將會整理我學習fastapi的一些例子，比如寫一個支持上傳文件的接口，post請求接受header中的，post請求，get請求等

1. 上傳文件
   接收上傳的文件，需要要io.BytesIO在內存中讀取二進制數據，通常圖像，音頻都會使用io.BytesIO。

import io
from scipy.io import wavfile
from fastapi import Body, FastAPI
import uvicorn 
from fastapi.responses import StreamingResponse
from PIL import Image
from fastapi import File
from fastapi.staticfiles import StaticFilesapp = FastAPI()@app.post("/v1")
async def ttt2image(file: bytes = File(...)): # 讀取原始音頻fs, sig_ori = wavfile.read(io.BytesIO(file))sig_ori = sig_ori.astype(float)# 去直流、歸一化sig_centered = sig_ori - np.mean(sig_ori)if np.max(np.abs(sig_centered)) != 0:sig_norm = sig_centered / np.max(np.abs(sig_centered))else:sig_norm = sig_centeredimage = Image.fromarray(image)img_byte_arr = io.BytesIO()image.save(img_byte_arr, format='PNG')image_base64 = base64.b64encode(img_byte_arr.getvalue()).decode('utf-8')return {"image": image_base64}if __name__ == '__main__':uvicorn.run(app='main:app', host="0.0.0.0", port=8135, reload=True, debug=False)

在命令行輸入
uvicorn main:app --reload --port 8005 --host 0.0.0.0
其中
–reload 是實時加載更新代碼
–port 是端口號
–host 默認是127.0.0.1 即localhost，指定0.0.0.0 時是實際ip地址
可以通過postman 進行接口請求測試了

2. post和get 示例

import io
from scipy.io import wavfile
from fastapi import Body, FastAPI
import uvicorn 
from fastapi.responses import StreamingResponse
from PIL import Image
from fastapi import File
from fastapi.staticfiles import StaticFilesapp = FastAPI()# post示例
@app.post("/v1")
async def ttt2image(file: bytes = File(...)): # 讀取原始音頻fs, sig_ori = wavfile.read(io.BytesIO(file))sig_ori = sig_ori.astype(float)image = Image.fromarray(image)img_byte_arr = io.BytesIO()image.save(img_byte_arr, format='PNG')image_base64 = base64.b64encode(img_byte_arr.getvalue()).decode('utf-8')return {"image": image_base64}@app.get("/logs")
async def logs(id: str=Query(...)):try:response = requests.get(url, params={"id": id})return response.json()except Exception as e:logger.info("message:{}".format(str(e)))return Response(content={"message": str(e)})if __name__ == '__main__':uvicorn.run(app='main:app', host="0.0.0.0", port=8135, reload=True, debug=False)

3. 子路由
   如果接口又很多全部寫在main.py中，不放便維護整理，需要把接口寫入子路由中，main.py 直接引入就可以，方便接口管理和維護。詳細看下面的例子。
   子路由的代碼util.py

from fastapi import Body
from fastapi import Query
from fastapi import Header
from fastapi import APIRouter
from fastapi import HTTPException
from fastapi import Response
from pydantic import BaseModelrouter = APIRouter()@router.post("/add")
async def add(items: Dict = Body(None)):value1 = items["value1"]value2 = items["value2"]result = (value1 + value2)* 4return {"respose": result}

主函數main.py

import asyncio
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from util import routerapp = FastAPI()# 接口前面加了api
app.include_router(router, prefix="/api")# 如果出現跨域的現象加上 下面幾行代碼
app.add_middleware(CORSMiddleware,allow_origins=["http://localhost:8005"],  # 在生產環境中應該限制具體的域名allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)async def main():config = uvicorn.Config(app,host="127.0.0.1",port=8006,log_level="info",access_log=False  )server = uvicorn.Server(config)await server.serve()if __name__ == "__main__":asyncio.run(main())# 等價于 下面的寫法
# if __name__ == '__main__':#     uvicorn.run(app='main:app', host="0.0.0.0", port=8135,  reload=True, debug=False)

以上是我最近工作中遇到的fastapi 內容，整理出來方便后續繼續更新，如果有寫的不合理的地方，歡迎指證。