Add configurable request duration logging middleware

docs_src/middleware/request_logging.py

@@ -0,0 +1,25 @@
+import logging
+from fastapi import FastAPI
+from fastapi.middleware.requestlogging import add_request_logging_middleware
+from fastapi.middleware.settings import RequestLoggingSettings
+
+logging.basicConfig(level=logging.INFO)
+
+app = FastAPI()
+
+add_request_logging_middleware(app)
+
+settings = RequestLoggingSettings(
+    request_logging_enabled=True,
+    request_logging_logger_name="my_app.requests",
+    request_logging_level=logging.DEBUG
+)
+add_request_logging_middleware(app, settings=settings)
+
+@app.get("/")
+async def read_root():
+    return {"Hello": "World"}
+
+@app.get("/items/{item_id}")
+async def read_item(item_id: int):
+    return {"item_id": item_id}

fastapi/middleware/__init__.py

@@ -1 +1,3 @@
 from starlette.middleware import Middleware as Middleware
+
+from .requestlogging import RequestLoggingMiddleware as RequestLoggingMiddleware

fastapi/middleware/requestlogging.py

@@ -0,0 +1,61 @@
+import logging
+import time
+from typing import Callable
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import Response
+
+
+class RequestLoggingMiddleware(BaseHTTPMiddleware):
+    def __init__(
+        self,
+        app,
+        logger_name: str = "fastapi.requests",
+        log_level: int = logging.INFO,
+        enabled: bool = True,
+    ):
+        super().__init__(app)
+        self.logger = logging.getLogger(logger_name)
+        self.log_level = log_level
+        self.enabled = enabled
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        if not self.enabled:
+            return await call_next(request)
+
+        start_time = time.perf_counter()
+        response = await call_next(request)
+        process_time = time.perf_counter() - start_time
+
+        self.logger.log(
+            self.log_level,
+            f"{request.method} {request.url.path} - {response.status_code} - {process_time:.4f}s"
+        )
+
+        return response
+
+
+def add_request_logging_middleware(
+    app,
+    settings=None,
+    **kwargs
+):
+    """
+    Add request duration logging middleware to a FastAPI app.
+
+    Args:
+        app: FastAPI application instance
+        settings: RequestLoggingSettings instance, or None to use defaults
+        **kwargs: Override specific settings
+    """
+    if settings is None:
+        from .settings import RequestLoggingSettings
+        settings = RequestLoggingSettings(**kwargs)
+
+    app.add_middleware(
+        RequestLoggingMiddleware,
+        logger_name=settings.request_logging_logger_name,
+        log_level=settings.request_logging_level,
+        enabled=settings.request_logging_enabled,
+    )

fastapi/middleware/settings.py

@@ -0,0 +1,11 @@
+import logging
+from pydantic import ConfigDict
+from pydantic_settings import BaseSettings
+
+
+class RequestLoggingSettings(BaseSettings):
+    model_config = ConfigDict(env_prefix="FASTAPI_")
+
+    request_logging_enabled: bool = True
+    request_logging_logger_name: str = "fastapi.requests"
+    request_logging_level: int = logging.INFO

tests/test_middleware_request_logging.py

@@ -0,0 +1,145 @@
+import logging
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from fastapi.middleware.requestlogging import RequestLoggingMiddleware, add_request_logging_middleware
+from fastapi.middleware.settings import RequestLoggingSettings
+
+
+def test_request_logging_middleware_basic():
+    app = FastAPI()
+
+    @app.get("/test")
+    async def test_endpoint():
+        return {"message": "test"}
+
+    app.add_middleware(RequestLoggingMiddleware)
+
+    client = TestClient(app)
+    response = client.get("/test")
+
+    assert response.status_code == 200
+    assert response.json() == {"message": "test"}
+
+
+def test_request_logging_middleware_with_custom_logger(caplog):
+    app = FastAPI()
+
+    @app.get("/test")
+    async def test_endpoint():
+        return {"message": "test"}
+
+    app.add_middleware(
+        RequestLoggingMiddleware,
+        logger_name="test.logger",
+        log_level=logging.INFO
+    )
+
+    client = TestClient(app)
+
+    with caplog.at_level(logging.INFO, logger="test.logger"):
+        response = client.get("/test")
+
+    assert response.status_code == 200
+    assert len(caplog.records) == 1
+    assert "GET /test - 200" in caplog.records[0].message
+    assert "s" in caplog.records[0].message
+
+
+def test_request_logging_middleware_disabled():
+    app = FastAPI()
+
+    @app.get("/test")
+    async def test_endpoint():
+        return {"message": "test"}
+
+    app.add_middleware(
+        RequestLoggingMiddleware,
+        enabled=False
+    )
+
+    client = TestClient(app)
+    response = client.get("/test")
+
+    assert response.status_code == 200
+    assert response.json() == {"message": "test"}
+
+
+def test_request_logging_settings():
+    settings = RequestLoggingSettings()
+    assert settings.request_logging_enabled is True
+    assert settings.request_logging_logger_name == "fastapi.requests"
+    assert settings.request_logging_level == logging.INFO
+
+
+def test_add_request_logging_middleware_convenience_function(caplog):
+    app = FastAPI()
+
+    @app.get("/test")
+    async def test_endpoint():
+        return {"message": "test"}
+
+    add_request_logging_middleware(app)
+
+    client = TestClient(app)
+
+    with caplog.at_level(logging.INFO, logger="fastapi.requests"):
+        response = client.get("/test")
+
+    assert response.status_code == 200
+    assert len(caplog.records) == 1
+    assert "GET /test - 200" in caplog.records[0].message
+
+
+def test_request_logging_middleware_different_methods(caplog):
+    app = FastAPI()
+
+    @app.get("/test")
+    async def get_endpoint():
+        return {"method": "GET"}
+
+    @app.post("/test")
+    async def post_endpoint():
+        return {"method": "POST"}
+
+    app.add_middleware(
+        RequestLoggingMiddleware,
+        logger_name="test.methods",
+        log_level=logging.INFO
+    )
+
+    client = TestClient(app)
+
+    with caplog.at_level(logging.INFO, logger="test.methods"):
+        get_response = client.get("/test")
+        post_response = client.post("/test")
+
+    assert get_response.status_code == 200
+    assert post_response.status_code == 200
+    assert len(caplog.records) == 2
+    assert "GET /test - 200" in caplog.records[0].message
+    assert "POST /test - 200" in caplog.records[1].message
+
+
+def test_request_logging_middleware_with_error_status(caplog):
+    app = FastAPI()
+
+    @app.get("/error")
+    async def error_endpoint():
+        from fastapi import HTTPException
+        raise HTTPException(status_code=404, detail="Not found")
+
+    app.add_middleware(
+        RequestLoggingMiddleware,
+        logger_name="test.errors",
+        log_level=logging.INFO
+    )
+
+    client = TestClient(app)
+
+    with caplog.at_level(logging.INFO, logger="test.errors"):
+        response = client.get("/error")
+
+    assert response.status_code == 404
+    assert len(caplog.records) == 1
+    assert "GET /error - 404" in caplog.records[0].message