Войти Регистрация
Конкурентность

33. Проверка статусов 500 ПВЗ Wildberries на asyncio

Лёгкая

Python · 32 строк

Раз в 5 минут сервис проверяет 500 ПВЗ своего региона: открыт ли пункт, сколько ячеек свободно. Один запрос к API статусов — около 200 мс, при перегрузке API отвечает 429.

В том же цикле событий живёт HTTP-хендлер /health. Оркестратор перезапускает под, если тот не отвечает 10 секунд. Отметьте строки с ошибками.

В работе: Фоновый сервис мониторинга пунктов выдачи.

Как проверяем Нажмите на строку с ошибкой и коротко напишите, что не так. Баг засчитан, если отмечена хотя бы одна его строка. Зачёт — найдено не меньше 60 % багов с учётом серьёзности (критичный весит 3, обычный 2, мелкий 1), а лишних отметок не больше, чем багов в коде. Пустые строки, скобки и комментарии лишними не считаем. Плюс 2 вопроса про поведение программы — на зачёт не влияют. Ответ один на подход: после него показываем разбор.
asyncioблокировка event looprequeststime.sleep

Источник: наша задача на данных Wildberries: пункты выдачи заказов. Условие пересказано своими словами.

  1. import asyncio
  2. import time
  4. import requests
  6. STATUS_URL = "https://pvz-api.internal/v1/points/{}/status"
  9. async def check_point(point_id: int) -> dict:
  10. for attempt in range(3):
  11. resp = requests.get(STATUS_URL.format(point_id), timeout=5)
  12. if resp.status_code != 429:
  13. return resp.json()
  14. time.sleep(1)
  15. raise RuntimeError(f"ПВЗ {point_id}: 429 три раза подряд")
  18. async def check_all(point_ids: list[int]) -> list[dict]:
  19. results = []
  20. for pid in point_ids:
  21. results.append(await check_point(pid))
  22. return results
  25. async def main():
  26. ids = list(range(1, 501))
  27. statuses = await check_all(ids)
  28. print(len(statuses))
  31. if __name__ == "__main__":
  32. asyncio.run(main())