Ревью Python-кода: типизация и идиомы

Резюме

Код почти не работает в продакшене. Синхронный httpx.get блокирует event loop. Первый же сбой поставщика завершает main_loop, и менеджер об этом не узнаёт: при ошибке уведомление не отправляется, а при успехе пишется в локальный файл, а не в Telegram. Главный бизнес-риск: except: pass молча выкидывает из отчёта заказы без промокода, а since_id считается по отфильтрованному списку, поэтому заказы теряются или обрабатываются повторно.

Замечания

Критично

1. fetch_orders: блокирующий вызов в корутине, client не используется httpx.get синхронный. Пока он ждёт ответа, весь event loop стоит, а timeout=10 и настройки клиента не применяются.

# до
resp = httpx.get(f"https://supplier.example.com/api/orders?since={since_id}")
# после
resp = await client.get(SUPPLIER_URL, params={"since": since_id})

2. process_batch / main_loop: ошибка поставщика убивает сервис, уведомления нет

• Исключение пробрасывается в while True без обработки, и после первого таймаута цикл завершается навсегда.
• Вложенный try/raise/except/raise e ничего не делает. Нужен raise ... from e, иначе причина видна только через неявный __context__.
• По ТЗ уведомление отправляется при ошибке. В коде оно отправляется при успехе, и в файл.

# до
except Exception:
    try:
        raise RuntimeError("Не удалось получить заказы от поставщика")
    except RuntimeError as e:
        raise e
# после
except (httpx.HTTPError, ValidationError) as e:
    await notify_manager(client, cfg, f"Ошибка получения заказов: {type(e).__name__}")
    raise SupplierError("Не удалось получить заказы от поставщика") from e

В main_loop исключение нужно ловить, логировать и продолжать цикл.

3. Голый except: pass в цикле заказов

• Заказ без promo_code или с неизвестным кодом вызывает KeyError и молча пропадает из отчёта.
• Голый except ловит и asyncio.CancelledError, то есть мешает корректной отмене задачи.

# до
try:
    processed = apply_discount(o)
    result.append(processed)
except:
    pass
# после: отсутствие промокода — штатная ветка, а не исключение
processed = [apply_discount(o, promos) for o in orders]

4. main_loop: since_id берётся из отфильтрованного result Если последние заказы батча упали в except, since_id не сдвигается, и они обрабатываются снова. Если упали все, батч повторяется бесконечно.

# до
since_id = batch[-1]["id"]
# после: считаем по исходным заказам
return max(o.id for o in orders)

5. asyncio.create_task(notify_manager(...)) без ссылки Event loop держит на задачу только слабую ссылку, поэтому её может собрать GC до завершения. Исключение из неё тоже теряется. Внутри при этом блокирующий open без with. Уведомление нужно просто дождаться через await, а если оно действительно фоновое, хранить ссылку в set и снимать её через add_done_callback.

Важно

6. float для денег. total * pct / 100 во float даёт значения вроде 899.9999999. Нужен Decimal и quantize(Decimal("0.01")).

7. apply_discount мутирует входной order. Лучше возвращать новый объект:

# до
order["final_price"] = discounted
return order
# после
return ProcessedOrder(id=order.id, final_price=total.quantize(CENT))

8. load_promocodes(path, extra_codes=[]): изменяемый дефолт. Список общий для всех вызовов и копится между ними. Параметр к тому же не нужен: функция должна возвращать словарь, а не заполнять глобальный объект.

9. Файлы без контекстного менеджера (out, f в notify_manager). При исключении в writerow дескриптор утекает, а CSV остаётся недописанным. Синхронная запись внутри корутины блокирует loop, её стоит вынести в asyncio.to_thread.

10. load_promocodes падает на заголовке и пустой строке (ValueError при распаковке). Нет encoding. Разбор через split(",") вместо csv.reader ломается на кавычках.

11. Типизация отсутствует полностью.

• Ответ API — непроверенный Any. Pydantic уже в зависимостях, поэтому стоит описать Order(BaseModel) и валидировать ответ через TypeAdapter(list[Order]).
• promo_code: str | None нужно проверять явно, а не через KeyError.
• PROMO_CACHE лучше заменить на Mapping[str, Decimal], передаваемый параметром.
• Сигнатуры: fetch_orders(client: httpx.AsyncClient, since_id: int) -> list[Order], process_batch(...) -> int.

12. promo == None → promo is None.

Желательно

13. for i in range(len(orders)): o = orders[i] заменить на списковое включение. writerows вместо цикла writerow.
14. Пути: "reports/" + str(since_id) + "/report.csv" заменить на REPORTS_DIR / str(since_id) / "report.csv", а os.makedirs на path.parent.mkdir(parents=True, exist_ok=True).
15. Query-параметры передавать через params=, а не f-строкой в URL.
16. logging.basicConfig выполняется при импорте модуля, это побочный эффект. Его место в main(). log объявлен, но не используется. FastAPI в зависимостях есть, но в коде не задействован.
17. Один AsyncClient стоит создавать на всё время жизни сервиса, а не на каждый батч: так переиспользуется пул соединений.

Итоговая версия

Самый проблемный фрагмент — получение заказов, обработка и основной цикл:

import asyncio
import csv
import logging
import os
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from decimal import ROUND_HALF_UP, Decimal
from pathlib import Path

import httpx
from pydantic import BaseModel, TypeAdapter, ValidationError

log = logging.getLogger("orders")

SUPPLIER_URL = "https://supplier.example.com/api/orders"
REPORTS_DIR = Path("reports")
POLL_INTERVAL_S = 600
CENT = Decimal("0.01")


@dataclass(frozen=True, slots=True)
class Config:
    tg_token: str
    tg_chat_id: str


class Order(BaseModel):
    id: int
    price: Decimal
    qty: int
    promo_code: str | None = None


class ProcessedOrder(BaseModel):
    id: int
    final_price: Decimal


class SupplierError(RuntimeError):
    pass


_orders_adapter = TypeAdapter(list[Order])


def load_promocodes(path: Path) -> dict[str, Decimal]:
    promos: dict[str, Decimal] = {}
    with path.open(encoding="utf-8", newline="") as f:
        for row in csv.reader(f):
            if not row:
                continue
            code, pct = row
            promos[code.strip()] = Decimal(pct.strip())  # мусор/заголовок -> падаем на старте
    return promos


async def fetch_orders(client: httpx.AsyncClient, since_id: int) -> list[Order]:
    resp = await client.get(SUPPLIER_URL, params={"since": since_id})
    resp.raise_for_status()
    return _orders_adapter.validate_json(resp.content)


def apply_discount(order: Order, promos: Mapping[str, Decimal]) -> ProcessedOrder:
    total = order.price * order.qty
    if order.promo_code is not None:
        pct = promos.get(order.promo_code)
        if pct is None:
            log.warning("Неизвестный промокод %r в заказе %s", order.promo_code, order.id)
        else:
            total -= total * pct / 100
    return ProcessedOrder(id=order.id, final_price=total.quantize(CENT, ROUND_HALF_UP))


def write_report(path: Path, rows: Iterable[ProcessedOrder]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["order_id", "final_price"])
        writer.writerows((r.id, r.final_price) for r in rows)


async def notify_manager(client: httpx.AsyncClient, cfg: Config, text: str) -> None:
    """Сбой уведомления не должен маскировать исходную ошибку."""
    try:
        resp = await client.post(
            f"https://api.telegram.org/bot{cfg.tg_token}/sendMessage",
            json={"chat_id": cfg.tg_chat_id, "text": text},
        )
        resp.raise_for_status()
    except httpx.HTTPError as e:
        # без exc_info: текст исключения содержит URL с токеном
        log.error("Не удалось отправить уведомление в Telegram: %s", type(e).__name__)


async def process_batch(
    client: httpx.AsyncClient,
    cfg: Config,
    promos: Mapping[str, Decimal],
    since_id: int,
) -> int:
    """Обрабатывает батч и возвращает новый since_id."""
    try:
        orders = await fetch_orders(client, since_id)
    except (httpx.HTTPError, ValidationError) as e:
        await notify_manager(client, cfg, f"Ошибка получения заказов (since={since_id}): {type(e).__name__}")
        raise SupplierError("Не удалось получить заказы от поставщика") from e

    if not orders:
        return since_id

    processed = [apply_discount(o, promos) for o in orders]
    await asyncio.to_thread(write_report, REPORTS_DIR / str(since_id) / "report.csv", processed)
    log.info("Обработано заказов: %d", len(processed))
    return max(o.id for o in orders)


async def main_loop(cfg: Config, promos: Mapping[str, Decimal]) -> None:
    since_id = 0
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            try:
                since_id = await process_batch(client, cfg, promos, since_id)
            except SupplierError:
                log.exception("Батч since=%s пропущен", since_id)
            except Exception:  # супервизор демона: падение одного батча не останавливает сервис
                log.exception("Непредвиденная ошибка в батче since=%s", since_id)
            await asyncio.sleep(POLL_INTERVAL_S)


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    cfg = Config(tg_token=os.environ["TG_BOT_TOKEN"], tg_chat_id=os.environ["TG_CHAT_ID"])
    promos = load_promocodes(Path("promocodes.csv"))
    asyncio.run(main_loop(cfg, promos))


if __name__ == "__main__":
    main()

Неизвестный промокод здесь означает полную цену и предупреждение в логе. Это бизнес-решение, его нужно согласовать: вариант с отправкой заказа на ручную проверку тоже разумен, но заказ в любом случае не должен молча пропадать.