in atr/routes/__init__.py [0:0]
def app_route_performance_measure(route_path: str, http_methods: list[str] | None = None) -> Callable:
"""Decorator that measures and logs route performance with path and method information."""
# def format_time(seconds: float) -> str:
# """Format time in appropriate units (µs or ms)."""
# microseconds = seconds * 1_000_000
# if microseconds < 1000:
# return f"{microseconds:.2f} µs"
# else:
# milliseconds = microseconds / 1000
# return f"{milliseconds:.2f} ms"
def decorator(f: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Awaitable[T]]:
@functools.wraps(f)
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
# This wrapper is based on an outstanding idea by Mostafa Farzán
# Farzán realised that we can step the event loop manually
# That way, we can also divide it into synchronous and asynchronous parts
# The synchronous part is done using coro.send(None)
# The asynchronous part is done using asyncio.sleep(0)
# We use two methods for measuring the async part, and take the largest
# This performance measurement adds a bit of overhead, about 10-20ms
# Therefore it should be avoided in production, or made more efficient
# We could perhaps use for a small portion of requests
blocking_time = 0.0
async_time = 0.0
loop_time = 0.0
total_start = time.perf_counter()
coro = f(*args, **kwargs)
try:
while True:
# Measure the synchronous part
sync_start = time.perf_counter()
future = coro.send(None)
sync_end = time.perf_counter()
blocking_time += sync_end - sync_start
# Measure the asynchronous part in two different ways
loop = asyncio.get_running_loop()
wait_start = time.perf_counter()
loop_start = loop.time()
if future is not None:
while not future.done():
await asyncio.sleep(0)
wait_end = time.perf_counter()
loop_end = loop.time()
async_time += wait_end - wait_start
loop_time += loop_end - loop_start
# Raise exception if any
# future.result()
except StopIteration as e:
total_end = time.perf_counter()
total_time = total_end - total_start
methods_str = ",".join(http_methods) if http_methods else "GET"
nonblocking_time = max(async_time, loop_time)
# If async time is more than 10% different from loop time, log it
delta_symbol = "="
nonblocking_delta = abs(async_time - loop_time)
# Must check that nonblocking_time is not 0 to avoid division by zero
if nonblocking_time and ((nonblocking_delta / nonblocking_time) > 0.1):
delta_symbol = "!"
route_logger.info(
"%s %s %s %s %s %s %s",
methods_str,
route_path,
f.__name__,
delta_symbol,
int(blocking_time * 1000),
int(nonblocking_time * 1000),
int(total_time * 1000),
)
return e.value
return wrapper
return decorator