Product Stat Importer

Мини-сервис на Django для импорта товарных данных из внешнего источника и выдачи агрегированной статистики через REST API.

• Python 3.12, Django 5, Django REST Framework
• PostgreSQL — хранилище товаров
• Pandas — нормализация входных данных
• Celery + Redis — периодический импорт по расписанию
• Redis (django-redis) — кэширование агрегатов
• Docker Compose — единая точка запуска всех сервисов
• pytest / pytest-django — тесты
• uv — управление зависимостями и виртуальным окружением

git clone https://github.com/Ste1l/product-stat-importer.git
cd product-stat-importer/server
cp .env.example .env
docker compose -f compose.dev.yml up -d --build

Поднимутся 5 сервисов: postgres, redis, web (Django + миграции), worker (Celery), beat (планировщик).

API будет доступен на http://localhost:8000.

Вариант 1 — вручную, через management command:

docker compose -f compose.dev.yml exec web uv run --no-sync python manage.py import_products

Вариант 2 — через Celery-таск напрямую:

docker compose -f compose.dev.yml exec web uv run --no-sync python manage.py shell -c \
  "from apps.catalog.tasks import import_products_task; import_products_task.delay()"

Вариант 3 — автоматически по расписанию: celery beat запускает импорт каждые IMPORT_INTERVAL_MINUTES минут (задаётся в .env, по умолчанию 15).

Используется публичный JSON API — FakeStore API.

Поля источника не совпадают с требуемой схемой, поэтому при импорте выполняется приведение:

• title → name
• updated_at в источнике отсутствует — проставляется момент импорта, если поле не пришло

5–10 строк пример�� ответа источника лежат в sample_data/sample_products.json — на случай, если внешний API будет временно недоступен.

Список товаров с фильтрами и пагинацией.

Параметры:

• category — точное совпадение категории
• price_min, price_max — диапазон цены
• page — номер страницы (пагинация по 20 записей)

Примеры:

curl "http://localhost:8000/api/items"
curl -G "http://localhost:8000/api/items" --data-urlencode "category=men's clothing"
curl "http://localhost:8000/api/items?price_min=50&price_max=200"
curl "http://localhost:8000/api/items?page=2"

Средняя цена товара по каждой категории. Результат кэшируется в Redis на 5 минут, кэш инвалидируется автоматически после каждого успешного импорта.

curl "http://localhost:8000/api/stats/avg-price-by-category"

Пример ответа:

[
  {"category": "electronics", "avg_price": 332.5},
  {"category": "jewelery", "avg_price": 221.0},
  {"category": "men's clothing", "avg_price": 51.06},
  {"category": "women's clothing", "avg_price": 26.29}
]

• REST через DRF, а не GraphQL — быстрее и надёжнее реализовать пагинацию, фильтры и корректные статус-коды с меньшим риском багов.
• Идемпотентность импорта построена на хэше содержимого записи (name + category + price), а не на сравнении updated_at. Так как источник (FakeStore API) не предоставляет собственную метку времени обновления, сравнение по updated_at привело бы к тому, что каждый повторный импорт считался бы "обновлением" (метка всегда новее предыдущей). Сравнение по хэшу гарантирует: если данные товара не изменились — запись помечается как skipped, а не перезаписывается.
• Естественный ключ идемпотентности — external_id = f"{name}__{category}", так как источник не отдаёт стабильный внешний ID товара, устойчивый между запросами.
• Кэш через django-redis напрямую (а не cacheops) — прозрачнее видно, что именно кэшируется и когда инвалидируется, вместо «магического» кэширования на уровне ORM.
• Инвалидация кэша происходит сразу после успешного upsert() в рамках run_import(), чтобы /stats/avg-price-by-category не отдавал устаревшие агрегаты дольше TTL.

docker compose -f compose.dev.yml exec web uv run --no-sync pytest

Покрытие:

• test_normalization.py — нормализация и приведение входных данных (переименование полей, отбрасывание невалидных строк, дедупликация, генерация updated_at, обработка ошибок схемы)
• test_stats.py — корректность расчёта средней цены по категории (агрегация и сам эндпоинт)
• test_api_filters.py — фильтрация по категории/цене и пагинация в /api/items

Тесты используют LocMemCache вместо реального Redis — независимы от внешней инфраструктуры.

server/ ├── compose.dev.yml ├── Dockerfile ├── pyproject.toml ├── manage.py ├── sample_data/ │ └── sample_products.json ├── config/ # settings, celery app, urls └── apps/ └── catalog/ ├── models.py # Item ├── serializers.py ├── views.py # ItemListView, AvgPriceByCategoryView ├── filters.py ├── tasks.py # Celery task ├── services/ │ └── importer.py # fetch + normalize (Pandas) + upsert ├── management/commands/import_products.py └── tests/