```python
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()
```
