1 如何快速實現一個fastapi

1.1 準備工作

使用Python來寫後端，基本上使用的是Django和Flask。fastapi不僅僅高效率而且還很適合產品級的開發。

需要安裝一下fastapi以及ASGI（ASGI是WSGI的升級版，支援非同步呼叫）

$ pip install fastapi
$ pip install pydantic
$ pip install uvicorn
$ pip install python-multipart

1.2 實現一個demo

1. 建立main.py, 然後寫入一下程式碼，可以直接執行

# -*- coding: utf-8 -*-
from fastapi import FastAPI
import uvicorn
# from web_server.views import router

app = FastAPI()


@app.get("/", tags=['首頁API'], description='fastapi後臺測試服務首頁API')
def index():
    return {'message': '指令碼後臺服務已啟動。', 'docs_url': 'http://127.0.0.1:8000/docs'}


@app.get("/items/{item_id}")
def read_item(item_id: int, keyword: str):
    """
    測試api
    :param item_id: url路徑引數
    :param keyword: url查詢字串引數
    :return: 
    """
    return {"item_id": item_id, "keyword": keyword}


# 新增子路由擴充套件API
# app.include_router(router, prefix='/exec_script')

if __name__ == '__main__':
    print('已啟動服務')
    # main指的是當前py檔名，app指的是定義的FastAPI()例項
    uvicorn.run(app='main:app', host='127.0.0.1', port=8000, reload=True, debug=True)
 

2. 因為安裝了uvicorn庫， 還可以通過終端命令執行：

$ uvicorn main:app --reload --port 8000
$ curl http://127.0.0.1:8000
$ curl http://127.0.0.1:8000/items/1?keyword=book

3. 響應的結果資料如下：

{
  "item_id": 1,
  "keyword": "book"
}

2 fastapi後臺非同步任務

2.1 後臺任務的使用

1. 建立一個非同步任務函式

   from fastapi import BackgroundTasks, FastAPI
   
   app = FastAPI()
   
   
   def write_notification(email: str, message=""):
       with open("log.txt", mode="w") as f:
           f.write(f"notification for {email}:{message}")
    
   

2. 呼叫後臺任務

   @app.post("/send-notification/{email}")
   async def send_notification(email: str, background_tasks: BackgroundTasks):
       background_tasks.add_task(write_notification, email, message="notification...")
       return {"message": "Notification sent in the background"}
   

   • add_task()接收的引數：

     • 一個在後臺執行的任務函式（write_notification）

     • 按照順尋傳遞的一系列引數（email）

     • 任何的關鍵字引數（message="notification..."）

2.2 依賴注入

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons