استفاده از middleware در FastAPI

مقدمه

Middleware در FastAPI چیست؟

اجازه دهید آنچه را که FastAPI رسمی در مورد Middleware توضیح داده است، تکرار کنیم.

“Middleware” تابعی است که با هر درخواستی قبل از پردازش توسط هر عملیات مسیر خاصی کار می کند. و همچنین با هر پاسخی قبل از بازگرداندن آن.

آزمایش 1: یک میان افزار ساده بسازید

بیایید مثال را در اسناد FastAPI امتحان کنیم. مثال اضافه کردن زمان پردازش API به سربرگ پاسخ است.

قبل از اضافه کردن میانافزاری که زمان پردازش را به سربرگ پاسخ اضافه می کند، بیایید یک نرم افزار ساده مانند زیر را امتحان کنیم.

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/")
async def root():
    return "Wonderful!!"

# by using curl, the http response header can be checked.
$ curl -i 127.0.0.1:8000
HTTP/1.1 200 OK
date: Sat, 12 Aug 2023 04:19:04 GMT
server: uvicorn
content-length: 13
content-type: application/json

"Wonderful!!"%    

بیایید میان افزار سفارشی را اضافه کنیم که “X-Process-Time” را به سربرگ پاسخ اضافه می کند.

برای ایجاد یک میان افزار، @app.middleware(“http”) در بالای یک تابع، add_process_time_header، اضافه می شود.

import time
from fastapi import FastAPI, Request

app = FastAPI()

# Implemented and added custom middleware to FastAPI
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start_time)
    return response

@app.get("/")
async def root():
    return "Wonderful!!"

ما می‌توانیم «x-process-time» را در هدر پاسخ مانند تصویر زیر ببینیم.

آزمایش 2: اگر مسیر منطبقی وجود نداشته باشد چه؟ آیا میان افزار اجرا می شود؟

من کنجکاو شدم که آیا میان افزار فقط در صورتی اجرا می شود که یک مسیر درخواست منطبق وجود داشته باشد.

ما یک پیام ساختگی اضافه کردیم که هر زمان که میان افزار اجرا شود چاپ می شود.

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    print("in add_process_time_header middleware.") # dummy message
    start_time = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start_time)
    return response

@app.get("/")
async def root():
    return "Wonderful!!"

ما دو درخواست با مسیر معتبر ارائه کردیم (شما می توانید دو “200 OK” را ببینید)

و، دو درخواست با مسیر نامعتبر ارائه کردیم (دو مورد “404 یافت نشد”)

بر اساس این آزمایش، توانستیم تأیید کنیم که میان افزار بدون توجه به وجود یا نبودن مسیر درخواست منطبق، اجرا می شود.

آزمایش 3: آیا میان افزار حتی برای نقطه پایانی غیر همگام کار می کند؟

اگر نقطه پایانی API غیر همگام باشد، میان افزار کار می کند؟ بله. کار می کند!

میان افزار را می توان برای نقاط انتهایی API همگام و غیر همگام استفاده کرد.

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)
    response.headers["X-Process-Time"] = str(time.time() - start_time)
    return response

@app.get("/")
def root():
    return "Wonderful!! - Sync"

آزمایش 4: ترتیب اعدام ها چگونه است؟

async def add_process_time_header(request: Request, call_next):
    # 1. Do thing before the matched path operation
    start_time = time.time()
    # 2. find / execute the matched path operation
    response = await call_next(request)
    # 3. Do thing after the matched path operation
    response.headers["X-Process-Time"] = str(time.time() - start_time)
    return response

FastAPI (دقیقاً Starlette) از پشته های میان افزار از جمله میان افزار کاربر عبور می کند. (میان افزار از پیش تعریف شده و از پیش اضافه شده + میان افزار کاربر وجود دارد که “add_process_time_header” است)

هنگامی که میان افزار «add_process_time_header» اجرا می شود، start_time = time.time() را اجرا می کند.

سپس، در میان‌افزار «add_process_time_header»، پاسخ = await call_next (درخواست) اجرا می‌شود. و مسیر همسان، مسیر def root() اجرا خواهد شد.

سپس، در میان‌افزار «add_process_time_header»، answer.headers[“X-Process-Time”] = str(time.time()-start_time) اجرا می‌شود.

بعد، پاسخ را برگردانید

FYI: کد Starlette که پشته میان افزار را می سازد

به این صورت است که پشته میان افزار ساخته می شود و برنامه با میان افزارها پیچیده می شود.

# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py

def build_middleware_stack(self) -> ASGIApp:
    debug = self.debug
    error_handler = None
    exception_handlers: typing.Dict[
        typing.Any, typing.Callable[[Request, Exception], Response]
    ] = {}

    for key, value in self.exception_handlers.items():
        if key in (500, Exception):
            error_handler = value
        else:
            exception_handlers[key] = value

    middleware = (
        [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
        + self.user_middleware
        + [
            Middleware(
                ExceptionMiddleware, handlers=exception_handlers, debug=debug
            )
        ]
    )

    app = self.router
    for cls, options in reversed(middleware):
        app = cls(app=app, **options)
    return app

آزمایش 5: ASGI Middlewares

Starlette (FastAPI مبتنی بر) بر اساس مشخصات ASGI پیاده سازی شده است.

این بدان معنی است که هر میان افزار ASGI (حتی از جمله شخص ثالث) را می توان در FastAPI نیز استفاده کرد.

ASGI Middlewares کلاس هایی هستند که انتظار دارند یک برنامه ASGI را به عنوان اولین آرگومان دریافت کنند.

در اینجا یک نمونه ASGI Middleware در Starlette آورده شده است. – GzipMiddleware

# ref: https://github.com/encode/starlette/blob/master/starlette/middleware/gzip.py
class GZipMiddleware:
    def __init__(
        self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9
    ) -> None:
        self.app = app
        self.minimum_size = minimum_size
        self.compresslevel = compresslevel

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http":
            headers = Headers(scope=scope)
            if "gzip" in headers.get("Accept-Encoding", ""):
                responder = GZipResponder(
                    self.app, self.minimum_size, compresslevel=self.compresslevel
                )
                await responder(scope, receive, send)
                return
        await self.app(scope, receive, send)

این میان افزارهای ASGI را می توان به برنامه FastAPI اضافه کرد، اما باید با استفاده از @app.middleware (“http”) کمی متفاوت باشد.

در اینجا مثالی آورده شده است که نشان می دهد چگونه ASGI Middleware — CORSMiddleware را می توان به برنامه FastAPI اضافه کرد.

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

add_middleware عملکرد Starlette در برنامه است. کلاس میان افزار داده شده را به لیست self.user_middleware اضافه می کند. کلاس میان‌افزار اضافه‌شده هنگام ساختن پشته میان‌افزار و بسته‌بندی برنامه استفاده می‌شود.

# https://github.com/encode/starlette/blob/master/starlette/applications.py
def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
    if self.middleware_stack is not None:  # pragma: no cover
        raise RuntimeError("Cannot add middleware after an application has started")
    self.user_middleware.insert(0, Middleware(middleware_class, **options))

آشنایی با فرمت add_middleware

app.add_middleware() یک کلاس میان افزار را به عنوان اولین آرگومان دریافت می کند و هر آرگومان اضافی را که باید به میان افزار ارسال شود.

app.add_middleware(
    CORSMiddleware, # middleware class
    # these are the keyword arguments for CORSMiddleware
    allow_origins=origins, 
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

در اینجا آرگومان های کلیدواژه init CORSMiddleware است.

# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py
class CORSMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        allow_origins: typing.Sequence[str] = (),
        allow_methods: typing.Sequence[str] = ("GET",),
        allow_headers: typing.Sequence[str] = (),
        allow_credentials: bool = False,
        allow_origin_regex: typing.Optional[str] = None,
        expose_headers: typing.Sequence[str] = (),
        max_age: int = 600,
    ) -> None:

بیشتر Middleware های موجود در FastAPI از Starlette سرچشمه می گیرند.

آزمایش 5: میان افزار سفارشی چگونه اضافه می شود؟

در مثال قبلی، با استفاده از app.middleware(“http”) یک میان افزار سفارشی اضافه کردم. این شبیه به ASGI Middleware نیست.

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start_time)
    return response

بیایید نگاهی به تابع، FastAPI middleware() بیندازیم.

در FastAPI، میان افزار مانند زیر پیاده سازی می شود. اساساً میان‌افزار سفارشی (فرمت میان‌افزار غیر ASGI) به عنوان بخشی از تابع اعزام BaseHTTPMiddleware اضافه می‌شود.

این یک راه ساده برای پیاده سازی و افزودن یک میان افزار سفارشی بدون پیاده سازی میان افزار کامل ASGI از ابتدا است.

# ref: https://github.com/tiangolo/fastapi/blob/master/fastapi/applications.py

def middleware(
    self, middleware_type: str
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        # this add_middleware is in Starlette
        # This is the format of adding ASGI middleware as you saw before.
        # BaseHTTPMiddleware is defined in Starlette
        # ref: https://github.com/encode/starlette/blob/master/starlette/middleware/base.py
        self.add_middleware(BaseHTTPMiddleware, dispatch=func)
        return func

    return decorator

میان افزار سفارشی که توسط BaseHTTPMiddleware تاب خورده است به self.user_middleware اضافه می شود.

بعداً از میان افزارها در self.user_middleware برای ساخت برنامه کامل استفاده می شود.

# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py

def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
    if self.middleware_stack is not None:  # pragma: no cover
        raise RuntimeError("Cannot add middleware after an application has started")
    self.user_middleware.insert(0, Middleware(middleware_class, **options))

اضافی: آیا می توان Middleware را فقط به روتر خاصی اضافه کرد؟

خیر، در حال حاضر (11–08–2023)، هنوز پشتیبانی نمی‌شود. (دقیقاً، هیچ راه تمیزی وجود ندارد، اما راه‌حل‌هایی وجود دارد.)

یک PR باز (در حال انجام) توسط نویسنده FastAPI وجود دارد که می تواند از ویژگی افزودن میان افزار در سطح روتر پشتیبانی کند.

اضافی: راه حل های استفاده از Middleware فقط در روتر خاص چیست؟

دو راه برای دستیابی به افزودن (یا فقط تأثیرگذاری) میان‌افزار در سطح روتر وجود دارد.

ساخت یک میان افزار سفارشی ASGI که از طریق مسیر فیلتر می کند.

اضافی: چگونه بدنه پاسخ را در Middleware تغییر دهیم؟

فکر کردم خیلی ساده است، اما به نظر نمی رسد.

این پست پیوندی نحوه پیاده‌سازی ASGI Middleware را نشان می‌دهد که بدنه پاسخ را تغییر می‌دهد.

منبع مطلب