import logging import tracemalloc from fastapi import Request # Start tracing at import time (once per process) tracemalloc.start() def log_memory_snapshot(request: Request | None = None): """Take a memory snapshot and log top 5 memory consumers. Optionally include request info for context. """ snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics("lineno") logging.info("[MEMORY SNAPSHOT] Top 5 Memory Consumers:") for idx, stat in enumerate(top_stats[:5], start=1): logging.info( f"#{idx}: {stat.traceback.format()[-1].filename}:{stat.traceback.format()[-1].lineno} - {stat.size / 1024:.2f} KiB" ) if request: logging.info(f"Request path: {request.url.path}")