Advanced Capstone Project

# -*- coding: utf-8 -*-

import asyncio
import logging
from datetime import datetime


class Producer:
    """Asynchronous producer that generates tasks"""

    def __init__(
        self,
        queue: asyncio.Queue,
        name: str,
        num_tasks: int = 5,
        delay: float = 0.5
    ):
        self.queue = queue
        self.name = name
        self.num_tasks = num_tasks
        self.delay = delay
        self.logger = logging.getLogger(self.name)

    async def produce(self) -> None:
        """Generate and enqueue tasks asynchronously"""
        self.logger.info(f"Producer {self.name} starting...")

        for i in range(self.num_tasks):
            task_data = {
                "id": f"{self.name}-task-{i}",
                "timestamp": datetime.now().isoformat(),
                "data": f"Data from {self.name}"
            }

            await self.queue.put(task_data)
            self.logger.info(
                f"Producer {self.name} produced: {task_data['id']}"
            )

            # Simulate work with async sleep
            await asyncio.sleep(self.delay)

        self.logger.info(f"Producer {self.name} finished")

# -*- coding: utf-8 -*-

import asyncio
import logging
from typing import Dict, Any


class Consumer:
    """Asynchronous consumer that processes tasks from queue"""

    def __init__(
        self,
        queue: asyncio.Queue,
        name: str,
        processing_time: float = 1.0
    ):
        self.queue = queue
        self.name = name
        self.processing_time = processing_time
        self.logger = logging.getLogger(self.name)
        self.tasks_processed = 0

    async def consume(self) -> None:
        """Consume and process tasks from the queue"""
        self.logger.info(f"Consumer {self.name} starting...")

        while True:
            try:
                # Wait for a task with timeout
                task_data = await asyncio.wait_for(
                    self.queue.get(),
                    timeout=3.0
                )

                # Process the task
                await self._process_task(task_data)

                # Mark task as done
                self.queue.task_done()
                self.tasks_processed += 1

            except asyncio.TimeoutError:
                # No more tasks in queue, exit
                self.logger.info(
                    f"Consumer {self.name} finished. "
                    f"Processed {self.tasks_processed} tasks"
                )
                break
            except Exception as error:
                self.logger.error(
                    f"Consumer {self.name} error: {error}"
                )

    async def _process_task(self, task_data: Dict[str, Any]) -> None:
        """Process a single task"""
        self.logger.info(
            f"Consumer {self.name} processing: {task_data['id']}"
        )

        # Simulate processing time
        await asyncio.sleep(self.processing_time)

        self.logger.info(
            f"Consumer {self.name} completed: {task_data['id']}"
        )

# -*- coding: utf-8 -*-

import asyncio
import logging
import os

from .producer import Producer
from .consumer import Consumer


async def execute() -> None:
    """Main async execution orchestrator"""
    logger = logging.getLogger(
        name=os.getenv("APP_NAME", "AsyncProducerConsumer")
    )

    try:
        logger.info("Starting async producer-consumer system...")

        # Create shared queue with max size
        queue = asyncio.Queue(maxsize=10)

        # Create multiple producers
        producers = [
            Producer(queue, name=f"Producer-{i}", num_tasks=5, delay=0.3)
            for i in range(3)
        ]

        # Create multiple consumers
        consumers = [
            Consumer(queue, name=f"Consumer-{i}", processing_time=0.8)
            for i in range(2)
        ]

        # Run all producers and consumers concurrently
        await asyncio.gather(
            *[p.produce() for p in producers],
            *[c.consume() for c in consumers]
        )

        # Wait for all tasks to be processed
        await queue.join()

        logger.info("All tasks completed successfully!")

    except Exception as error:
        logger.error(f"System error: {error}")
        raise

    finally:
        logger.info("Async system shutdown complete")

# -*- coding: utf-8 -*-

import asyncio
import logging
import os

from click.core import CommandCollection
from click.decorators import group
from environs import Env


# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)


@group()
def cli_main():
    pass


@cli_main.command("run")
def main():
    """Run the async producer-consumer system"""
    from service.main import execute

    # Run async code from sync context
    asyncio.run(execute())
    exit()


if __name__ == "__main__":
    Env().read_env(os.path.join(os.getcwd(), ".env"))
    CommandCollection(sources=[cli_main])()