Python asyncio: обработка и фиксация пачек данных

Условие

import asyncio
from typing import List, Optional

class Chunk:
    def __init__(self, items: List[dict]):
        self.items = items

    @property
    def size(self) -> int:
        return len(self.items)

class Producer:
    async def next(self) -> Optional[Chunk]:
        # возвращает пачку данных или None
        ...

    async def commit(self, chunks: List[Chunk]) -> None:
        # фиксируем успешную обработку chunks
        ...

class Consumer:
    max_batch_size = 100

    async def consume(self, chunks: List[Chunk]) -> None:
        total_size = sum(chunk.size for chunk in chunks)

        if total_size > self.max_batch_size:
            raise ValueError("batch is too large")

        # обработка данных
        ...

producer = Producer()
consumer = Consumer()

async def process():
    while True:
        chunk = await producer.next()

        if chunk is None:
            break

        try:
            await consumer.consume([chunk])
            await producer.commit([chunk])
        except Exception as e:
            print("error but continue:", e)
            continue

if __name__ == "__main__":
    asyncio.run(process())

Ответ

В представленном коде нужно реализовать методы Producer.next(), Producer.commit() и логику обработки в Consumer.consume().

• Producer.next() должен асинхронно возвращать следующий объект Chunk с данными или None, если данных больше нет.
• Consumer.consume() проверяет, что суммарный размер пачки не превышает max_batch_size, и обрабатывает данные (например, сохраняет или анализирует).
• Producer.commit() фиксирует успешную обработку пачки, например, обновляя статус в базе.

Пример реализации с имитацией данных:

import asyncio
from typing import List, Optional

class Chunk:
    def __init__(self, items: List[dict]):
        self.items = items

    @property
    def size(self) -> int:
        return len(self.items)

class Producer:
    def __init__(self):
        self.data = [
            [{'id': i} for i in range(30)],
            [{'id': i} for i in range(50)],
            [{'id': i} for i in range(120)],  # слишком большой чанк
            [{'id': i} for i in range(20)],
        ]
        self.index = 0

    async def next(self) -> Optional[Chunk]:
        await asyncio.sleep(0.1)  # имитация задержки
        if self.index >= len(self.data):
            return None
        chunk = Chunk(self.data[self.index])
        self.index += 1
        return chunk

    async def commit(self, chunks: List[Chunk]) -> None:
        await asyncio.sleep(0.05)  # имитация фиксации
        print(f"Committed {sum(chunk.size for chunk in chunks)} items")

class Consumer:
    max_batch_size = 100

    async def consume(self, chunks: List[Chunk]) -> None:
        total_size = sum(chunk.size for chunk in chunks)
        if total_size > self.max_batch_size:
            raise ValueError(f"batch is too large: {total_size} > {self.max_batch_size}")
        await asyncio.sleep(0.1)  # имитация обработки
        print(f"Consumed {total_size} items")

producer = Producer()
consumer = Consumer()

async def process():
    while True:
        chunk = await producer.next()
        if chunk is None:
            break
        try:
            await consumer.consume([chunk])
            await producer.commit([chunk])
        except Exception as e:
            print("error but continue:", e)
            continue

if __name__ == "__main__":
    asyncio.run(process())

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