Импорт заказов: незапущенная корутина и скрытые ошибки БД

Условие

import logging
import asyncio
import json
import sqlite3
from datetime import datetime
from typing import Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OrderImporter:
    """
    Сервис импорта заказов из внешней системы в локальную базу.
    """

    def __init__(self, db_path: str, api_client):
        """
        Инициализирует импортёр
        """
        self.db_path = db_path
        self.api_client = api_client
        self.connection = sqlite3.connect(db_path)
        self.cache = {}

    async def import_orders(
        self,
        customer_ids: list[int],
        imported_orders: list[dict[str, Any]] = [],
    ) -> dict[str, Any]:
        """ Импортирует заказы для списка клиентов """
        results = {
            "created": 0,
            "failed": 0,
            "orders": imported_orders,
        }

        for customer_id in customer_ids:
            try:
                # Внешний API возвращает список сырых заказов клиента.
                orders = self.api_client.get_orders(customer_id)

                for order in orders:
                    normalized_order = await self.normalize_order(order)

                    # Отменённые заказы не должны попадать в БД
                    if normalized_order["status"] == "cancelled":
                        continue

                    await self.save_order(normalized_order)

                    results["created"] += 1
                    results["orders"].append(normalized_order)

            except Exception as error:
                logger.error(
                    f"Cannot import orders for customer {customer_id}: {error}"
                )
                results["failed"] += 1
        return results

    async def normalize_order(self, order: dict[str, Any]) -> dict[str, Any]:
        """
        Преобразует заказ из формата внешнего API во внутренний формат.

        Внешняя система может возвращать дополнительные поля, поэтому
        импортёр сохраняет только данные, необходимые для локального сервиса.

        Args:
            order: Исходный заказ, полученный из API.

        Returns:
            Нормализованный словарь с данными заказа.
        """
        order_id = order.get("id")
        total = float(order.get("total", 0))
        items = order.get("items", [])

        # Имитация дополнительной асинхронной обработки заказа,
        # например получения данных из другого сервиса.
        await asyncio.sleep(0.1)

        return {
            "id": order_id,
            "customer_id": order["customer"]["id"],
            "email": order["customer"]["email"].lower(),
            "total": total,
            "items": items,
            "items_count": len(items),
            "status": order.get("status", "new"),
            # Время, когда заказ был обработан импортёром.
            "created_at": datetime.now().isoformat(),
        }

    async def save_order(self, order: dict[str, Any]) -> None:
        """
        Сохраняет нормализованный заказ в БД
        Перед сохранением проверяет локальный кэш, чтобы не выполнять
        повторную запись одного и того же заказа.
        """
        if order["id"] in self.cache:
            return

        # Заказ хранится в одной таблице.
        # Состав заказа сериализуется в JSON для сохранения в SQLite.
        query = """
            INSERT INTO orders (
                id,
                customer_id,
                email,
                total,
                items,
                status,
                created_at
            ) VALUES (
                ?,
                ?,
                ?,
                ?,
                ?,
                ?,
                ?
            )
        """

        try:
            cursor = self.connection.cursor()
            cursor.execute(query, (
                order["id"],
                order["customer_id"],
                order["email"],
                order["total"],
                json.dumps(order["items"]),
                order["status"],
                order["created_at"]
            ))

            # Добавляем заказ в кэш только после выполнения SQL-запроса.
            self.cache[order["id"]] = order
            self.connection.commit()

        except sqlite3.Error:
            logger.exception("Database error")

        finally:
            cursor.close()

    def get_order(self, order_id: int) -> dict[str, Any]:
        """ Возвращает заказ по его идентификатору """
        cursor = self.connection.cursor()

        result = cursor.execute(
            f"SELECT * FROM orders WHERE id = {order_id}"
        ).fetchone()

        if not result:
            return {}

        return {
            "id": result[0],
            "customer_id": result[1],
            "email": result[2],
            "total": result[3],
            "items": result[4],
            "status": result[5],
            "created_at": result[6],
        }

    def close(self):
        self.connection.close()

async def main():
    """ Запускает импорт заказов для списка клиентов """
    importer = OrderImporter(
        db_path="orders.db",
        api_client=ExternalOrdersApiClient(),
    )

    result = await importer.import_orders([1, 2, 3])

    print(f"Imported: {result['created']}")
    print(f"Failed: {result['failed']}")

    importer.close()

if __name__ == "__main__":
    main()

Ответ

Программа не запускает корутину main, а ошибки БД маскируются под успешный импорт. Дополнительно нужны корректные транзакции, идемпотентность и валидация заказов.

В точке входа нужен asyncio.run(main()); показанный ExternalOrdersApiClient также должен быть определён или импортирован. Замените изменяемый аргумент [] на None.

save_order глотает ошибку, поэтому вызывающий код увеличивает created даже при неудаче или попадании в кеш. Возвращайте явный результат, выполняйте rollback и обновляйте кеш только после commit. Защитите id уникальным ограничением БД.

Синхронные API/SQLite блокируют event loop; выбирайте подходящий клиент или контролируемое выполнение вне него. Деньги храните без float, проверяйте поля, используйте UTC. В get_order нужны параметры SQL, закрытие cursor и json.loads(items). Закрытие соединения вынесите в finally; определите, failed считает клиентов или заказы.