Created
March 11, 2024 17:52
-
-
Save rajaravivarma-r/bb9bd10e9c21084410918061d113582c to your computer and use it in GitHub Desktop.
FASTApi with logging in middleware
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import logging | |
| import asyncio | |
| import json | |
| from asyncio import sleep | |
| from fastapi import FastAPI, APIRouter, Request, File, UploadFile | |
| from fastapi.routing import APIRoute | |
| from fastapi.encoders import jsonable_encoder | |
| from fastapi.responses import ( | |
| FileResponse, | |
| JSONResponse, | |
| Response, | |
| StreamingResponse, | |
| HTMLResponse, | |
| ) | |
| from starlette.background import BackgroundTask | |
| def log_info(msg): | |
| logging.info(msg) | |
| logging.info("=" * 100) | |
| class LoggingRoute(APIRoute): | |
| def get_route_handler(self): | |
| original_route_handler = super().get_route_handler() | |
| async def custom_route_handler(request: Request) -> Response: | |
| request_body_size = 0 | |
| response_body = b"" | |
| response_status = None | |
| text_response = True | |
| request_body_string = "<BINARY_CONTENT>" | |
| new_response = None | |
| request_headers = request.headers | |
| request_body = await request.body() | |
| response = await original_route_handler(request) | |
| if isinstance(response, StreamingResponse): | |
| async for item in response.body_iterator: | |
| response_body += item | |
| new_response = Response( | |
| content=response_body, | |
| status_code=response.status_code, | |
| headers=dict(response.headers), | |
| media_type=response.media_type, | |
| background=task, | |
| ) | |
| else: | |
| if hasattr(response, "body"): | |
| response_body = response.body | |
| new_response = response | |
| elif isinstance(response, FileResponse): | |
| response_body = f"<File:{response.path}".encode() | |
| new_response = response | |
| try: | |
| request_body_string = request_body.decode() | |
| except UnicodeDecodeError as e: | |
| log_info(f"Exception decoding request body: {e}") | |
| msg = json.dumps( | |
| { | |
| "request": { | |
| "body": request_body_string, | |
| "size": request_body_size, | |
| "path": str(request.url), | |
| "query_string": dict(request.query_params), | |
| "method": request.method, | |
| "headers": dict(request_headers), | |
| }, | |
| "response": { | |
| "body": response_body.decode(), | |
| "status_code": response_status, | |
| "headers": dict(response.headers), | |
| }, | |
| }, | |
| indent=4, | |
| ) | |
| response.background = BackgroundTask(log_info, msg) | |
| return new_response | |
| return custom_route_handler | |
| app = FastAPI(debug=True) | |
| router = APIRouter(route_class=LoggingRoute) | |
| logging.basicConfig(filename="info.log", level=logging.DEBUG) | |
| @router.get("/home") | |
| async def home(): | |
| html_content = public_index_html_path.read_text() | |
| return HTMLResponse(content=html_content, status_code=200) | |
| @router.api_route( | |
| "/{full_path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"] | |
| ) | |
| async def fallback_routes(request: Request, full_path: str): | |
| return JSONResponse({"message": "ok"}, status_code=200) | |
| app.include_router(router) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment