-
Notifications
You must be signed in to change notification settings - Fork 3
/
middleware.py
43 lines (31 loc) · 1.16 KB
/
middleware.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# Standard Library
import re
import time
# Third Party
from loguru import logger
from starlette.middleware.base import BaseHTTPMiddleware
@logger.catch
class StatsMiddleware(BaseHTTPMiddleware):
"""Get duration of request"""
async def dispatch(self, request, call_next):
url = str(request.url)
method = str(request.method)
t0 = time.time()
response = await call_next(request)
url = url.rstrip("/")
route_name = url.split("/")[-1]
# logger.info("Skipping status/metrics", url=url, route_name=route_name)
if method == "OPTIONS" or route_name in ["status", "metrics", "ping"]:
return response
else:
t1 = time.time()
duration = f"{(t1 - t0) * 1000:.0f}"
logger.opt(exception=True).info(
"Request Metrics {duration_ms} ms, status_code: {status_code} {method} {url}",
# "Request Metrics {duration_ms} ms, status_code: {status_code}",
duration_ms=duration,
status_code=response.status_code,
method=method,
url=url,
)
return response