Существует ли способ FastAPI для доступа к текущим данным запроса в глобальном масштабе?

Question

В рамках FastAPI:

Хотя данные запроса, безусловно, могут быть переданы в качестве аргумента, я хотел бы знать, может ли функция получить доступ к информации о текущем запросе без передачи аргумента.

Отказ от ответственности: я не считаю глобальный доступ к запросу данных хорошей практикой, и тем не менее у меня есть пример использования, при котором было бы очень неплохо это сделать.

Answer 1

Предлагаемое здесь решение определяет диспетчер контекста, к которому вы можете получить доступ глобально. Для каждого запроса вы извлекаете соответствующую информацию (например, заголовки) и передаете ее диспетчеру контекста.

Поскольку fastapi построен на Starlette, вы можете использовать библиотеку starlette-context. Это создаетcontextобъект, который можно использовать, не передавая его в качестве аргумента. Главное предостережение заключается в том, что вам все равно нужно передавать объект запроса на все ваши маршруты.

Пример кода из этой библиотеки для иллюстрации:

import uvicorn
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import JSONResponse

from starlette_context import context, plugins
from starlette_context.middleware import ContextMiddleware


app = FastAPI(debug=True)
app.add_middleware(ContextMiddleware.with_plugins(  # easily extensible
    plugins.RequestIdPlugin,  # request id
    plugins.CorrelationIdPlugin  # correlation id
))


@app.route('/')
async def index(request: Request):  # This argument is still needed here
    return JSONResponse(context.data)  # Your context data


uvicorn.run(app, host="0.0.0.0")

Answer 2

Хотел предоставить обновленный ответ здесь. Мой оригинальный комментарий относится к проблеме звездочки, здесь .

Я использую FastAPI, и мне нужен способ доступа к информации об объекте запроса вне представления. Сначала я рассматривал использование starlette-context, но нашел следующее решение, которое подходит для моих нужд.

Спасибо Марку (см. вопрос о звездочке выше) за основу этого решения.

Как отметил выше Колин Ле Ност, авторы предостерегают от использованияBaseHTTPMiddleware-- родительский класс, от которого наследуется промежуточное ПО Марка.

Вместо этого предлагается использовать необработанное промежуточное ПО ASGI. Однако для этого не так много документации. Я смог использовать AuthenticationMiddleware от Starlette в качестве ориентира и разработать то, что мне было нужно, в сочетании с замечательным решением Marc ContextVars.

      # middleware.py
from starlette.types import ASGIApp, Receive, Scope, Send

REQUEST_ID_CTX_KEY = "request_id"

_request_id_ctx_var: ContextVar[str] = ContextVar(REQUEST_ID_CTX_KEY, default=None)

def get_request_id() -> str:
    return _request_id_ctx_var.get()

class CustomRequestMiddleware:
    def __init__(
        self,
        app: ASGIApp,
    ) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in ["http", "websocket"]:
            await self.app(scope, receive, send)
            return

        request_id = _request_id_ctx_var.set(str(uuid4()))

        await self.app(scope, receive, send)

        _request_id_ctx_var.reset(request_id)

И затем в настройках приложения:

      # main.py
app.add_middleware(CustomRequestMiddleware)

И, наконец, функция non-view:

      # myfunc.py
import get_request_id

request_id = get_request_id()

Это должно позволить вам использовать ContextVars как способ получить любую информацию от объекта запроса, который вам нужен, и сделать его доступным вне функции просмотра. Еще раз спасибо всем в этой теме за всю помощь, и я надеюсь, что вышеизложенное полезно!

Answer 3

Вы можете получить / установить произвольные атрибуты на request.state

Пожалуйста, обратитесь к проблеме ниже для подробного объяснения и реализации:

https://github.com/tiangolo/fastapi/issues/633

Answer 4

Я обычно делал это, используя очередь сообщений в стиле производитель-потребитель. У меня есть пример репозитория, показывающий, как я использую глобальную очередь для передачи данных из почтового запроса в WebSocket, который передает это клиентам.

Хотя это может не быть вашим точным вариантом использования, вы сможете адаптировать его под себя.

По сути, это класс Notifier, который помещает данные в очередь:

async def push(self, msg: str):
    await self.channel.default_exchange.publish(
        Message(msg.encode("ascii")),
        routing_key=self.queue_name,
    )

Что касается потребителя, у меня есть _notify функция, которая принимает сообщения из очереди и отправляет их через WebSocket:

async def _notify(self, message: IncomingMessage):
    living_connections = []
    while len(self.connections) > 0:
        websocket = self.connections.pop()
        await websocket.send_text(f"{message.body}")
        living_connections.append(websocket)
    self.connections = living_connections

Answer 5

Вы можете использовать Starlette Request

например:

from starlette.requests import Request
from fastapi import FastApi

app = FastApi()
@app.get('/')
def get(request:Request):
    requests_header = request.headers
    return "Hi"