FastAPI学習の準備–5週間のミッドレンジ実施、ログの記録

私の主な活動のコミュニティHOLIXの開発者の中で異なっている開発者と話をします!
https://github.com/kanikim/fastapiに収録!

Middlewareとは?
ミドルウェアについては様々なコメントと定義がありますが、FastAPIでは以下の説明が提供されています.
A "middleware"is a function that works with every request before it is processed by any specific path operation. And also with every response before returning it.
ミドルウェアは、path操作が完了する前にすべてのリクエストを処理し、応答を返す前に処理する関数です.
すなわち,ここでFastAPIは仲介者として働き,ミドルウェアリクエストを受信する前に何かを処理し,responseに戻る前に何かを処理する.では、これらの中間バッグを本当に使う場所はどこなのか考えてみましょう.まず思いついたのはログを撮ることそこで、今日の記事ではFastAPIでログを撮影してみます.
FastAPI Middleware
FastAPI Middleware Flow
FastAPIのミドルウェアには、次のフローがあります.
Request

は、まず各要求をアプリケーションに送信する.

以降、この要求は、作成されたコードによって処理される.

の結果を残りのアプリケーションコードに移行します.

Response

アプリケーションによって生成されたリースを受け入れます.

の後、このリースを持って処理します.

はその後、Response

に戻る
上記の単純なプロセスで実行されます.では、FastAPIでMiddlewareを作成するにはどうすればいいのでしょうか.とても簡単です.FastAPIが提供するdecoratorを使用できます.
FastAPI Middleware Decorator

import time

from fastapi import FastAPI, Request

app = FastAPI()


@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

上のコードで@app.midelware(「http」)セクションは、ミドルウェアとして指定されたセクションです.上のコードをよく見てみましょう.最も重要なパラメータ部分

async def add_process_time_header(request: Request, call_next):

requestには、実際の要求としてAPIに送信される情報が含まれる.では、隣のcall nextは何ですか?call nextは、API要求をAPIに対応するパスに送信し、応答を返す関数パラメータです.この2つのパラメータは、カスタムミドルウェアを作成する際に使用することを推奨します.
FastAPI logging
では、簡単なプログラムを正式に作成し、ログを記録しましょう.完全なコードは次のとおりです.

import random
import uuid
import contextvars
from loguru import logger
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

request_id_contextvar = contextvars.ContextVar("request_id", default=None)

def divide(a, b):
    print(f"Dividing {a} / {b} ... ")
    result = a / b
    print(f"Result is {result}")


@app.middleware("http")
async def request_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request_id_contextvar.set(request_id)
    logger.debug("Request Started")

    try:
        return await call_next(request)

    except Exception as e:
        logger.debug(f"Request failed: {e}")
        return JSONResponse(content={"Success": False}, status_code=500)

    finally:
        assert request_id_contextvar.get() == request_id
        logger.debug("Request ended")


@app.get("/")
def read_root():
    a = 100
    b = random.randint(0, 1)
    return {"success": True, "resut": divide(a, b)}

今回の学習では、他のパッケージも使用します.contextvarsは、loguruに相当します.一つ一つ見てみましょう.
contextvars
contextvars-コンテキスト変数は、ローカル状態およびアクセスを管理および格納するためのAPIです.コードでは、

contextvars.ContextVar("request_id", default=None)

request_id_contextvar.set(request_id)

request_id_contextvar.get()

一緒に使う部分.それぞれ説明するには、まずcontextvarsです.変数をContextVarで宣言します.名前は「request id」です.次にsetでrequest idを使用して新しい値を設定します.次にgetで現在のコンテキスト変数の値を返します.では、これらのcontextvarはいつ使われたのでしょうか.

import contextvars

def show():
    print('value is', val.get())

val = contextvars.ContextVar("val", default=0)
contexts = list()

for _ in range(5):
    ctx = contextvars.copy_context()
    ctx.run(val.set, val.get()+1)

    contexts.append(ctx)

for ctx in contexts:
    ctx.run(show)

上のコードを見せてください.コードのみからctxは1、2、3、4、5を出力する可能性がありますが、出力結果はいずれも1です.これはvalを増やすcontextとfor loopを実行するcontextが異なるcontextであるため独立して存在する.つまり、他のコンテキストでは異なる値を使用できます.
非同期タスクの他のタスクで異なる値を保持する必要がある場合は、この値を使用します.
loguru
loguruはPythonベースのログ記録オープンソースで、機能が強く、設定する必要がなく使用でき、非同期タスクもサポートされています.簡単で使いやすい:

from loguru import logger
logger.debug("hello world")

ぶんかいコード
コードを見てみましょう.

app = FastAPI()

request_id_contextvar = contextvars.ContextVar("request_id", default=None)

def divide(a, b):
    print(f"Dividing {a} / {b} ... ")
    result = a / b
    print(f"Result is {result}")

このコードは、除算をログに残し、上記のdivide(a,b)関数で値を取得してa/b操作を実行するコードです.aとbは整数のみを渡し、bはExceptionで使用するために0を渡します.これは簡単なアプリケーション実行コードであり、最も重要な部分である@appです.meddlware(「http」)セクションを見てみましょう.

@app.middleware("http")
async def request_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request_id_contextvar.set(request_id)
    logger.debug("Request Started")

    try:
        return await call_next(request)

    except Exception as e:
        logger.debug(f"Request failed: {e}")
        return JSONResponse(content={"Success": False}, status_code=500)

    finally:
        assert request_id_contextvar.get() == request_id
        logger.debug("Request ended")

まずrequest idのuuidです.uuid 4()を設定します.自分のプライマリ・キーを入力するだけでログを切断できます.前述したようにcontext varを設定します.その後loguruでデバッグします.最初に実行される画面は次のとおりです.

2022-03-01 10:11:33.128 | DEBUG | main:request_middleware:28 - Request Started
印刷が見える.
次に、try構文を使用してawait call next(request)を最初に実行します.このとき実行されるコードは@appです.get("/")セクション.このコードは、上記のプロセスで説明したアプリケーションセグメントに対応します.コードはa=100、b=0または1である.これにより、Exceptionを自然に強制的に実行することができます.
したがって、Exceptionを起動すると、ログは次のように印刷されます.

よく見るとログがよく撮れているのが見えます.
Summary
この文章で、簡単にログを撮ってみました.Middlewareではリクエストを前処理したりスポンサーを前処理したりできます.