विषय पर बढ़ें

Custom Request और APIRoute class

🌐 एआई और मनुष्यों द्वारा किया गया अनुवाद

यह अनुवाद मनुष्यों के मार्गदर्शन में एआई द्वारा किया गया है। 🤝

इसमें मूल अर्थ को गलत समझने या अप्राकृतिक लगने आदि जैसी गलतियाँ हो सकती हैं। 🤖

आप हमें एआई LLM को बेहतर मार्गदर्शन करने में मदद करके इस अनुवाद को बेहतर बना सकते हैं।

अंग्रेज़ी संस्करण

कुछ मामलों में, आप Request और APIRoute classes द्वारा उपयोग किए जाने वाले logic को override करना चाह सकते हैं।

विशेष रूप से, यह middleware में logic का एक अच्छा विकल्प हो सकता है।

उदाहरण के लिए, अगर आप request body को आपके application द्वारा process किए जाने से पहले पढ़ना या manipulate करना चाहते हैं।

खतरा

यह एक "advanced" feature है।

अगर आप अभी-अभी FastAPI के साथ शुरुआत कर रहे हैं, तो शायद आप इस section को छोड़ना चाहेंगे।

Use cases

कुछ use cases में शामिल हैं:

  • non-JSON request bodies को JSON में convert करना (जैसे msgpack).
  • gzip-compressed request bodies को decompress करना।
  • सभी request bodies को automatically log करना।

Custom request body encodings को संभालना

आइए देखें कि gzip requests को decompress करने के लिए custom Request subclass का उपयोग कैसे करें।

और उस custom request class का उपयोग करने के लिए एक APIRoute subclass।

Custom GzipRequest class बनाएँ

सुझाव

यह दिखाने के लिए एक सरल उदाहरण है कि यह कैसे काम करता है, अगर आपको Gzip support चाहिए, तो आप दिए गए GzipMiddleware का उपयोग कर सकते हैं।

सबसे पहले, हम एक GzipRequest class बनाते हैं, जो उपयुक्त header की मौजूदगी में body को decompress करने के लिए Request.body() method को overwrite करेगी।

अगर header में gzip नहीं है, तो यह body को decompress करने की कोशिश नहीं करेगी।

इस तरह, वही route class gzip compressed या uncompressed requests को संभाल सकती है।

import gzip
from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

import gzip
from collections.abc import Callable

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: list[int] = Body()):
    return {"sum": sum(numbers)}

Custom GzipRoute class बनाएँ

इसके बाद, हम fastapi.routing.APIRoute की एक custom subclass बनाते हैं जो GzipRequest का उपयोग करेगी।

इस बार, यह APIRoute.get_route_handler() method को overwrite करेगी।

यह method एक function return करता है। और वही function एक request प्राप्त करेगा और response return करेगा।

यहाँ हम इसका उपयोग original request से GzipRequest बनाने के लिए करते हैं।

import gzip
from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

import gzip
from collections.abc import Callable

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: list[int] = Body()):
    return {"sum": sum(numbers)}

तकनीकी विवरण

एक Request में request.scope attribute होता है, जो बस एक Python dict है जिसमें request से संबंधित metadata होता है।

एक Request में request.receive भी होता है, जो request की body को "receive" करने के लिए एक function है।

scope dict और receive function दोनों ASGI specification का हिस्सा हैं।

और ये दो चीजें, scope और receive, नई Request instance बनाने के लिए आवश्यक हैं।

Request के बारे में अधिक जानने के लिए Requests के बारे में Starlette के docs देखें।

GzipRequest.get_route_handler द्वारा return किए गए function का केवल एक अलग काम है: Request को GzipRequest में convert करना।

ऐसा करने से, हमारा GzipRequest data को हमारे path operations तक पास करने से पहले decompress करने का ध्यान रखेगा (अगर आवश्यक हो)।

उसके बाद, processing logic सब वही रहता है।

लेकिन GzipRequest.body में हमारे बदलावों की वजह से, request body जरूरत पड़ने पर FastAPI द्वारा load किए जाने पर automatically decompress हो जाएगी।

Exception handler में request body तक पहुँचना

सुझाव

इसी समस्या को हल करने के लिए, RequestValidationError के custom handler में body का उपयोग करना शायद बहुत आसान है (Errors संभालना)।

लेकिन यह उदाहरण अभी भी valid है और यह दिखाता है कि internal components के साथ कैसे interact करें।

हम इसी approach का उपयोग exception handler में request body तक पहुँचने के लिए भी कर सकते हैं।

हमें बस request को try/except block के अंदर संभालना है:

from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return sum(numbers)
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from collections.abc import Callable

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: list[int] = Body()):
    return sum(numbers)

अगर कोई exception होता है, तो Request instance अभी भी scope में होगी, इसलिए error संभालते समय हम request body को पढ़ सकते हैं और उसका उपयोग कर सकते हैं:

from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return sum(numbers)
🤓 Other versions and variants

Tip

Prefer to use the Annotated version if possible.

from collections.abc import Callable

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: list[int] = Body()):
    return sum(numbers)

Router में custom APIRoute class

आप APIRouter का route_class parameter भी set कर सकते हैं:

import time
from collections.abc import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=TimedRoute)


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)

इस उदाहरण में, router के अंतर्गत path operations custom TimedRoute class का उपयोग करेंगे, और response में एक अतिरिक्त X-Response-Time header होगा जिसमें response generate करने में लगा समय होगा:

import time
from collections.abc import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=TimedRoute)


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)