python大型項目後臺異步這一篇就夠了

點擊python編程從入門到實踐，置頂公衆號重磅 python入門資料，第一時間送達python

讀完須要 6 分鐘編程

速讀僅需 2 分鐘api

/ 後臺異步這一篇就夠了 /服務器

1微信

BackgroundTasks 使用場景閉包

有時候咱們須要在 request 執行以後繼續一些操做，但終端並不須要等待這些操做完成才能收到 response 。我列舉一些場景你們看一下：併發

1.在自動出票完成後須要向各 ota 平臺自動發送行程單信息
2.在執行完購票後須要向各戶發送郵件通知購票成功信息
3.收到客戶端的文件以後對文件進行二次處理
4....
5....app

這些操做都須要必定的處理時間，但與返回給終端的 response 並沒有直接關係這個時候就能夠經過定義後臺任務 BackgroundTasks 來實現這個功能。異步

2async

BackgroundTasks 實戰

2.1

添加參數

首先咱們須要導入 BackgroundTasks，並在路徑操做函數中添加 BackgroundTasks 類型的參數。

# -*- encoding: utf-8 -*-
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    """
    發送提醒任務
    """
　　pass

2.2

任務函數

任務函數是指：在須要建立一個在後臺任務中實際執行的函數。任務函數是一個標準函數。這個函數能夠是 async def 或者普通 def 的函數。

這裏建立一個把指定內容寫入文件的函數。

# -*- encoding: utf-8 -*-
def write_notification(email: str, message=""):
    """
    寫入文件操做模擬任務
    """
    with open("log_test.txt", mode="w") as email_file:
        content = "notification for {}: {}".format(email,message)
        email_file.write(content)

2.3

添加後臺任務

最後須要把任務函數添加到後臺任務中

# -*- encoding: utf-8 -*-
import time
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    """
    寫入文件操做模擬任務
    """
    time.sleep(5)
    with open("log_test.txt", mode="w") as email_file:
        content = "notification for {}: {}".format(email,message)
        email_file.write(content)
    print("write end")
    time.sleep(2)


@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")

    return {"message": "now: %s Notification sent in the background" % time.time()}

2.4

add_task 解釋

.add_task()接收如下參數信息：1.在後臺運行的任務函數(例如 write_notification)2.任意序列的參數信息(例如 email)3.任意鍵值對的參數信息(例如 message="some notification")4.咱們故意在 write_notification 方法中加入等待時間來驗證對於客戶端的返回

2.5

依賴注入

後臺任務能夠與依賴注入系統一塊兒使用，能夠在不一樣層級的依賴項中聲明 BackgroundTasks 參數，若對依賴注入系統還不是很熟悉的小夥伴能夠會看一下上篇內容，本節就不作過多的解釋了

# -*- encoding: utf-8 -*-
from typing import Optional
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
    　　message = f"found query: {q}\n" background_tasks.add_task(write_log, message)
    return q


@app.post("/send_info")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)):

    message = f"message to {email}\n" background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

1.若須要執行大量的後臺計算而沒必要必定要在同一進程中運行，例如：它不須要共享內存，變量等，則可以使用其餘更大的工具，例如:celery、MQ 系列都是能夠選擇的但這些每每須要更復雜的配置，例如:RabbitMQ、Redis 之類的消息做業隊列管理器，可是它們容許在多個進程（尤爲是多個服務器）中運行後臺任務。
2.若須要從同一 FastAPI 應用訪問變量和對象，或者須要執行一些小的後臺任務例如:發送電子郵件、短信消息等，則只需使用便可 BackgroundTasks。

咱們啓動項目看一下執行效果：

我堅信：思考問題的方法遠大於具體解決問題的方案，讓咱們繼續一路前行，下期再見！

原創不易，只願能幫助那些須要這些內容的同行或剛入行的小夥伴，你的每次點贊、分享都是我繼續創做下去的動力，我但願能在推廣 python 技術的道路上盡我一份力量，歡迎在評論區向我提問，我都會一一解答，記得一鍵三連支持一下哦！

加入python學習交流微信羣，請後臺回覆「入羣」

往期推薦

python生產實戰 python 閉包之庖丁解牛篇

大型fastapi項目實戰靠 python 中間件解決方案漲薪了

大型fastapi項目實戰高併發請求神器之aiohttp(下)

大型fastapi項目實戰高併發請求神器之aiohttp(上) [建議收藏]

本文分享自微信公衆號 - python編程軍火庫（PythonCoder1024）。
若有侵權，請聯繫 support@oschina.cn 刪除。
本文參與「OSC源創計劃」，歡迎正在閱讀的你也加入，一塊兒分享。