api_memory_watch

Библиотека для профилирования потребления памяти FastAPI-эндпоинтов.

api_memory_watch позволяет навешивать декоратор на нужные маршруты и автоматически собирать метрики памяти во время обработки HTTP-запроса. Библиотека фиксирует:

• память процесса на уровне ОС (RSS);
• память Python-объектов через tracemalloc;
• пиковое потребление памяти во время запроса;
• удержанную память после завершения запроса;
• подозрение на утечку памяти по серии вызовов одного и того же эндпоинта;
• понятный человекочитаемый отчёт в лог-файл.

Для чего нужна библиотека

Библиотека полезна, когда нужно:

• понять, сколько памяти реально потребляет конкретный FastAPI-метод;
• найти эндпоинты с избыточным пиковым потреблением памяти;
• выявить маршруты, после которых память не возвращается к базовому уровню;
• локально воспроизводить и анализировать подозрения на memory leak;
• получать понятный отчёт без ручного запуска профилировщиков.

Что библиотека измеряет

1. RSS процесса

RSS (Resident Set Size) — это объём памяти процесса, который видит операционная система.

Используется для оценки реального потребления памяти приложением.

Плюсы:

• отражает общую память процесса;
• полезен для анализа поведения приложения в проде и на стендах;
• позволяет увидеть накопление памяти между вызовами.

Особенности:

• показатель может быть шумным;
• Python не всегда сразу возвращает память обратно ОС;
• рост RSS не всегда означает утечку.

2. Python memory через tracemalloc

tracemalloc показывает память, выделенную Python-объектами.

Плюсы:

• позволяет понять, сколько памяти удержано именно Python-кодом;
• показывает пиковое потребление памяти во время запроса;
• позволяет смотреть top allocations по файлам и строкам.

Как определяется подозрение на утечку

Библиотека не делает ложного утверждения, что утечка может быть доказана по одному вызову.

Логика такая:

• один вызов показывает, сколько памяти было выделено и сколько памяти удержалось после завершения запроса;
• серия одинаковых вызовов одного и того же эндпоинта формирует историю retained memory;
• если удержанная память устойчиво растёт на протяжении окна наблюдения, библиотека помечает эндпоинт как suspected_leak=true.

Иными словами, библиотека ищет не разовый всплеск памяти, а тренд накопления.

Архитектура

Библиотека строится на двух основных механизмах:

1. Декоратор-маркер

Декоратор @memory_watch(...) навешивается на конкретный endpoint и только помечает его для профилирования.

Он не выполняет тяжёлую логику сам по себе.

2. Кастомный APIRoute

Класс MemoryWatchRoute перехватывает обработку запроса и снимает метрики до и после вызова эндпоинта.

Это нужно, чтобы корректно учитывать:

• dependency injection FastAPI;
• сериализацию ответа;
• работу response_model;
• память, выделенную уже после возврата значения из функции.

Такой подход точнее обычного декоратора вокруг функции.

Структура библиотеки

api_memory_watch/
├── __init__.py
├── config.py
├── decorator.py
├── detector.py
├── route.py
└── sinks.py

Назначение файлов

• config.py — dataclass-конфигурация и модели метрик.
• decorator.py — декоратор @memory_watch(...).
• detector.py — эвристика определения подозрения на утечку.
• route.py — основной перехватчик FastAPI-маршрутов.
• sinks.py — логирование метрик в файл в человекочитаемом виде.
• __init__.py — публичные экспорты библиотеки.

Требования

• Python 3.10+
• FastAPI
• Uvicorn
• psutil

Установка

Если библиотека находится внутри вашего проекта, достаточно добавить файлы в репозиторий и установить зависимости.

pip install fastapi uvicorn psutil

tracemalloc входит в стандартную библиотеку Python и отдельно не устанавливается.

Быстрый старт

Шаг 1. Подключить библиотеку в main.py

from fastapi import FastAPI
from api_memory_watch import configure_memory_watch, MemoryWatchConfig
from routers.debug_memory import router as debug_memory_router

app = FastAPI()

configure_memory_watch(
    MemoryWatchConfig(
        enabled=True,
        use_tracemalloc=True,
        tracemalloc_frames=20,
        top_stats_limit=10,
        leak_window=5,
        leak_min_growth_bytes=1_000_000,
        leak_min_retained_bytes=300_000,
        logger_name="memory_watch",
        log_file_path="logs/memory_watch.log",
    )
)

app.include_router(debug_memory_router)

Шаг 2. Создать роутер с route_class=MemoryWatchRoute

from fastapi import APIRouter
from api_memory_watch import MemoryWatchRoute, memory_watch

router = APIRouter(
    prefix="/debug-memory",
    tags=["debug-memory"],
    route_class=MemoryWatchRoute,
)


@router.get("/ok")
@memory_watch(name="debug_memory.ok")
async def ok():
    return {"status": "ok"}

Шаг 3. Запустить приложение

uvicorn main:app --host 127.0.0.1 --port 8000 --workers 1

Для анализа памяти рекомендуется использовать именно один worker.

Почему нужен один worker

Если запустить несколько worker-процессов, метрики памяти будут смешиваться между разными процессами и интерпретировать результат станет существенно сложнее.

Для локальной диагностики рекомендуется:

uvicorn main:app --host 127.0.0.1 --port 8000 --workers 1

И желательно без --reload, так как авто-перезагрузка тоже создаёт шум по памяти.

Важно: оба условия обязательны

Чтобы метрики снимались, нужно выполнить оба условия:

1. у роутера должен быть route_class=MemoryWatchRoute;
2. на endpoint должен быть навешан @memory_watch(...).

Если указать только декоратор без MemoryWatchRoute, библиотека не будет перехватывать обработку запроса.

Формат логов

Библиотека пишет метрики в файл, например:

2026-03-09 19:13:45,415 | [debug_memory.heavy] GET /debug-memory/heavy -> 200
Время выполнения: 8.81 ms

Процесс до запроса занимал: 57.57 MB
Процесс после запроса занимал: 57.59 MB
Прирост памяти процесса: 0.02 MB

Python-объекты до запроса: 5.15 MB
Python-объекты после запроса: 5.15 MB
Удержанная память: 0.00 MB
Пиковое потребление во время запроса: 5.39 MB

Признаки утечки: не обнаружены
----------------------------------------------------------------------------------------------------

Как читать лог

Процесс до запроса занимал

Объём RSS процесса перед входом в обработку эндпоинта.

Процесс после запроса занимал

Объём RSS после завершения обработки запроса.

Прирост памяти процесса

Разница между RSS после и до запроса.

Python-объекты до запроса

Объём памяти Python-объектов перед выполнением эндпоинта.

Python-объекты после запроса

Объём памяти Python-объектов после выполнения эндпоинта.

Удержанная память

Память Python-объектов, оставшаяся после завершения запроса.

Пиковое потребление во время запроса

Максимальный объём Python memory, достигнутый внутри одного вызова.

Признаки утечки

Результат работы leak detector на основе истории вызовов одного и того же эндпоинта.

Когда это не утечка

Высокое пиковое потребление не обязательно означает leak.

Например, если у метода:

• высокий Пиковое потребление во время запроса;
• небольшой Удержанная память;
• Признаки утечки: не обнаружены;

это обычно означает тяжёлый, но не текущий endpoint, а не накопление памяти между вызовами.

Типичный пример — метод, который грузит большой список записей, сериализует их и затем освобождает память.

Когда есть серьёзное подозрение на leak

Если по серии вызовов одного и того же endpoint наблюдается:

• стабильно положительная удержанная память;
• рост retained memory от вызова к вызову;
• систематический рост baseline;
• Признаки утечки: обнаружены;

тогда это уже серьёзный сигнал, что объекты где-то удерживаются между запросами.

Пример тестового роутера для локальной проверки

from fastapi import APIRouter
from api_memory_watch import MemoryWatchRoute, memory_watch

router = APIRouter(
    prefix="/debug-memory",
    tags=["debug-memory"],
    route_class=MemoryWatchRoute,
)

LEAK_BUCKET = []


@router.get("/ok")
@memory_watch(name="debug_memory.ok")
async def ok():
    return {"status": "ok"}


@router.get("/heavy")
@memory_watch(name="debug_memory.heavy")
async def heavy(size_mb: int = 10):
    data = ["x" * 1024 for _ in range(size_mb * 1024)]
    return {"allocated_mb": size_mb, "items": len(data)}


@router.get("/leak")
@memory_watch(name="debug_memory.leak")
async def leak(size_mb: int = 5):
    chunk = "x" * (size_mb * 1024 * 1024)
    LEAK_BUCKET.append(chunk)
    return {
        "status": "leak_simulated",
        "size_mb": size_mb,
        "bucket_size": len(LEAK_BUCKET),
    }


@router.delete("/leak/reset")
async def reset_leak():
    LEAK_BUCKET.clear()
    return {"status": "reset"}

Сценарии локальной проверки

1. Проверка, что библиотека подключена

curl http://127.0.0.1:8000/debug-memory/ok

Ожидаемый результат:

• запрос отвечает 200;
• в файл лога записывается блок метрик.

2. Проверка большого потребления памяти без leak

curl "http://127.0.0.1:8000/debug-memory/heavy?size_mb=20"

Ожидаемый результат:

• заметно увеличивается пиковое потребление памяти;
• удержанная память остаётся относительно небольшой;
• признак утечки обычно не срабатывает.

3. Проверка подозрения на leak

Несколько раз подряд:

curl "http://127.0.0.1:8000/debug-memory/leak?size_mb=5"

Ожидаемый результат:

• по мере накопления LEAK_BUCKET растёт retained memory;
• после заполнения окна наблюдения endpoint может получить Признаки утечки: обнаружены.

Где смотреть лог-файл

Если в конфигурации указан путь:

log_file_path="logs/memory_watch.log"

то лог будет находиться здесь:

logs/memory_watch.log

Просмотр на Windows

Get-Content .\logs\memory_watch.log

Просмотр в реальном времени:

Get-Content .\logs\memory_watch.log -Wait

Просмотр на Linux

cat logs/memory_watch.log

В реальном времени:

tail -f logs/memory_watch.log

Конфигурация

MemoryWatchConfig

MemoryWatchConfig(
    enabled=True,
    use_tracemalloc=True,
    tracemalloc_frames=20,
    top_stats_limit=10,
    force_gc_before=False,
    force_gc_after=False,
    leak_window=5,
    leak_min_growth_bytes=1_000_000,
    leak_min_retained_bytes=300_000,
    logger_name="memory_watch",
    log_file_path="logs/memory_watch.log",
)

Параметры конфигурации

enabled

Глобальное включение и выключение библиотеки.

use_tracemalloc

Включает сбор Python memory через tracemalloc.

tracemalloc_frames

Сколько кадров traceback хранить для tracemalloc.

Чем выше значение, тем больше точность для анализа источников аллокаций, но тем выше overhead.

top_stats_limit

Сколько top allocations хранить в метрике.

force_gc_before

Вызывать ли gc.collect() до обработки запроса.

force_gc_after

Вызывать ли gc.collect() после обработки запроса.

Обычно держат False, чтобы не искажать естественное поведение приложения.

leak_window

Размер окна наблюдения по вызовам одного endpoint.

leak_min_growth_bytes

Минимальный суммарный рост retained memory, после которого можно подозревать утечку.

leak_min_retained_bytes

Минимальный retained memory, считающийся значимым.

logger_name

Имя логгера.

log_file_path

Путь к файлу логов.

Декоратор memory_watch

Пример:

@memory_watch(name="city.get_cities")

Параметры

name

Человекочитаемое имя endpoint, которое попадает в лог.

Если не указать, библиотека может использовать комбинацию HTTP method и path.

enabled

Локально выключает профилирование для конкретного endpoint.

capture_top_stats

Включает сбор top allocations для данного endpoint.

Практические рекомендации

1. Начинайте с локального тестового роутера

Сначала проверьте работу библиотеки на /debug-memory/ok, /debug-memory/heavy и /debug-memory/leak, а уже потом подключайте к реальным бизнес-эндпоинтам.

2. Не интерпретируйте один вызов как утечку

Один вызов даёт только моментальный снимок.

Для утечки нужен устойчивый тренд по серии однотипных запросов.

3. Проверяйте одинаковые входные параметры

Если вы хотите сравнивать поведение метода, серия вызовов должна быть максимально одинаковой.

4. Используйте один worker

Иначе метрики памяти будет сложно сопоставлять с конкретным вызовом.

5. Подключайте библиотеку выборочно

Не обязательно профилировать все endpoint одновременно.

Чаще всего достаточно подключить:

• тяжёлые списочные методы;
• экспорт/генерацию файлов;
• отчёты;
• методы с сериализацией больших объектов;
• подозрительные эндпоинты, после которых растёт память.

Продакшен-рекомендации

Dev / Stage

Рекомендуется:

• use_tracemalloc=True
• детализированные логи
• выборочная проверка тяжёлых endpoint

Prod

Рекомендуется осторожное использование:

• либо отключить tracemalloc, если нужен минимальный overhead;
• либо включать библиотеку только на ограниченном наборе endpoint;
• либо использовать профилирование временно во время диагностики.

Ограничения библиотеки

Важно понимать ограничения:

1. библиотека не может со 100% точностью доказать утечку памяти по одному вызову;
2. рост RSS не всегда означает leak;
3. Python allocator не всегда возвращает память ОС сразу;
4. tracemalloc видит только Python-аллокации, а не всю native memory;
5. при высокой конкурентности показатели становятся шумнее;
6. при нескольких worker-процессах интерпретация усложняется.

Типичные причины высокого потребления памяти

Часто проблема оказывается не в утечке, а в неэффективной логике endpoint.

Например, такой код:

cities = list(
    (await self.session.execute(select(City).order_by(City.date_create.desc())))
    .scalars()
    .all()
)

означает, что в память загружается вся таблица, а потом уже применяется пагинация в Python.

Это приводит к:

• высокому пику памяти;
• лишней сериализации;
• плохой масштабируемости.

В таком случае библиотека покажет проблему, но это будет не leak, а архитектурно тяжёлый endpoint.

Рекомендации по оптимизации

Если библиотека показала большой пик памяти, проверьте:

• не грузится ли вся таблица в память целиком;
• не выполняется ли пагинация после загрузки всех записей;
• нет ли больших списков, словарей или base64-строк в ответе;
• не удерживаются ли объекты в глобальных переменных, singleton-сервисах или кэше;
• не копируются ли большие структуры без необходимости.

Troubleshooting

Не появляются метрики

Проверьте:

• вызван ли configure_memory_watch(...);
• указан ли route_class=MemoryWatchRoute у роутера;
• навешан ли @memory_watch(...) на endpoint;
• корректен ли путь log_file_path;
• есть ли права на запись в директорию логов.

Ошибка ModuleNotFoundError: No module named 'psutil'

Установите зависимость:

python -m pip install psutil

Ошибка ImportError: attempted relative import with no known parent package

Используйте абсолютные импорты вместо запуска файла с относительными импортами.

Например:

from api_memory_watch import configure_memory_watch, MemoryWatchConfig

а запускать лучше так:

uvicorn main:app --host 127.0.0.1 --port 8000 --workers 1

Лог пустой или не создаётся

Проверьте:

• существует ли директория для логов;
• не заблокирован ли файл другой программой;
• не используется ли некорректный путь на Windows;
• не отключён ли логгер из-за повторного конфигурирования.

Публичный API библиотеки

from api_memory_watch import (
    MemoryWatchConfig,
    MemoryWatchRoute,
    configure_memory_watch,
    memory_watch,
)

Пример экспорта в __init__.py

from .config import MemoryWatchConfig
from .decorator import memory_watch
from .route import MemoryWatchRoute, configure_memory_watch

__all__ = [
    "MemoryWatchConfig",
    "MemoryWatchRoute",
    "configure_memory_watch",
    "memory_watch",
]

Дальнейшее развитие

Библиотеку можно расширить следующими возможностями:

• запись raw-метрик в JSON параллельно с человекочитаемым логом;
• экспорт метрик в Prometheus;
• отдельный alert-файл только для suspected_leak=true;
• фильтрация top allocations только по файлам проекта;
• сохранение истории в Redis или PostgreSQL;
• отдельные пороги утечки для разных endpoint.

Итог

api_memory_watch — это прикладная библиотека для FastAPI, которая помогает быстро и понятно ответить на три вопроса:

1. сколько памяти съедает конкретный endpoint;
2. какой пик памяти возникает во время обработки запроса;
3. есть ли признаки накопления памяти между одинаковыми вызовами.

Она хорошо подходит для локальной диагностики, stage-стендов и выборочного анализа тяжёлых endpoint в реальных сервисах.