FastAPI 学习笔记
最近在公司中需要写接口,选取了fastapi这个框架,一个原因是FastAPI 是主流框架,同时FastAPI 有着高性能,支持异步和高并发。
FastAPI 安装
直接用下面两行命令进行安装
pip3 install fastapi
pip install uvicorn
FastAPI 例子
下面将会整理我学习fastapi的一些例子,比如写一个支持上传文件的接口,post请求接受header中的,post请求,get请求等
- 上传文件
接收上传的文件,需要要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 进行接口请求测试了
- 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)
- 子路由
如果接口又很多全部写在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 内容,整理出来方便后续继续更新,如果有写的不合理的地方,欢迎指证。