/
urusai
/
code_sample
Обзор
Документация
Войти
/
urusai
/
code_sample
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/utils/logs/middlewares.py
62 строки
2 KB
nick
linters
30 сен 2024, 16:13
30 сен 2024, 16:13
09a0f4a
Код
Авторство
О чём код?
import logging from io import BytesIO from fastapi import Request, Response from starlette.middleware.base import RequestResponseEndpoint from utils.logs.json_logger import EMPTY_VALUE logger = logging.getLogger(__name__) class AsyncIteratorWrapper: """ Small helper to turn BytesIO into async-able iterator """ def __init__(self, bytes_: bytes): super().__init__() self._it = BytesIO(bytes_) def __aiter__(self): return self async def __anext__(self): try: value = next(self._it) except StopIteration: raise StopAsyncIteration return value class LoggingMiddleware: """ Middleware для обработки запросов и ответов с целью журналирования """ @staticmethod async def get_protocol(request: Request) -> str: protocol = str(request.scope.get("type", "")) http_version = str(request.scope.get("http_version", "")) if protocol.lower() == "http" and http_version: return f"{protocol.upper()}/{http_version}" return EMPTY_VALUE async def __call__( self, request: Request, call_next: RequestResponseEndpoint, *args, **kwargs ): message = ( f'{"Ошибка" if exception_object else "Ответ"} ' f"с кодом {response.status_code} " f'на запрос {request.method} "{str(request.url)}", ' f"за {duration} мс" ) logger.info( message, extra={ "request_json_fields": request_json_fields, "to_mask": True, }, exc_info=exception_object, ) return response