initial commit

README.md

@@ -1,12 +1,200 @@
----
-title: Async Framework
-emoji: 🐢
-colorFrom: red
-colorTo: gray
-sdk: docker
-pinned: false
-license: mit
-short_description: Distributed batch framework using python
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Async Framework
+
+A Python library for Redis-based message queue processing with monitoring and retry capabilities. This framework provides a robust solution for distributed message processing with features like automatic retries, dead letter queues, and Prometheus metrics.
+
+## Features
+
+- Redis-based message broker and consumer
+- Configurable number of workers
+- JSON message format with standardized structure
+- Efficient Redis connection pooling
+- Prometheus metrics for monitoring
+- Automatic and manual retry mechanisms
+- Dead letter queue for failed messages
+- Batch processing support
+- Distributed environment ready
+
+## Requirements
+
+- Python 3.10+
+- Redis server
+- Dependencies (automatically installed):
+  - redis>=4.5.0
+  - pydantic>=2.0.0
+  - prometheus-client>=0.17.0
+  - tenacity>=8.2.0
+  - structlog>=23.1.0
+
+## Installation
+
+### Installing from Source
+1. Clone the repository:
+```bash
+git clone https://github.com/yourusername/async-framework.git
+cd async-framework
+```
+
+2. Install the package:
+```bash
+pip install .
+```
+
+### Building the Package
+To build the package as a wheel file:
+
+1. Install build requirements:
+```bash
+pip install build wheel
+```
+
+2. Build the package:
+```bash
+python -m build
+```
+
+This will create two files in the `dist` directory:
+- `async_framework-0.1.0-py3-none-any.whl` (Wheel file)
+- `async_framework-0.1.0.tar.gz` (Source distribution)
+
+### Installing the Built Package
+
+You can install the wheel file in other projects using:
+```bash
+pip install async_framework-0.1.0-py3-none-any.whl
+```
+
+Or install directly from your private PyPI repository:
+```bash
+pip install async-framework
+```
+
+## Usage
+
+### Configuration
+
+```python
+from async_framework import BrokerConfig, RedisConfig, RetryConfig
+
+config = BrokerConfig(
+    redis=RedisConfig(
+        host="localhost",
+        port=6379,
+        password="optional_password",
+        connection_pool_size=10
+    ),
+    retry=RetryConfig(
+        max_retries=3,
+        initial_delay=1.0,
+        max_delay=60.0,
+        backoff_factor=2.0
+    ),
+    num_workers=4,
+    batch_size=10,
+    polling_interval=1.0,
+    metrics_port=8000
+)
+```
+
+### Publishing Messages
+
+```python
+from async_framework import MessageBroker
+
+# Initialize the broker
+broker = MessageBroker(config)
+
+# Publish a message
+message = broker.publish(
+    queue="my_queue",
+    payload={"key": "value"},
+    max_retries=3  # Optional, defaults to config value
+)
+```
+
+### Consuming Messages
+
+```python
+from async_framework import MessageConsumer, Message
+
+# Define a message handler
+def process_message(message: Message):
+    print(f"Processing message {message.id}: {message.payload}")
+    # Your processing logic here
+    # Raise an exception to trigger retry mechanism
+
+# Initialize the consumer
+consumer = MessageConsumer(
+    config=config,
+    queue="my_queue",
+    handler=process_message
+)
+
+# Start consuming messages
+consumer.start()
+
+try:
+    # Keep the main thread running
+    while True:
+        time.sleep(1)
+except KeyboardInterrupt:
+    consumer.stop()
+```
+
+### Monitoring
+
+The framework exposes Prometheus metrics at `http://localhost:8000` (configurable via `metrics_port`):
+
+- `messages_published_total` - Counter of published messages by queue
+- `messages_processed_total` - Counter of processed messages by queue and status
+- `message_processing_seconds` - Histogram of message processing time by queue
+
+### Queue Structure
+
+The framework uses the following Redis key patterns:
+
+- `queue:{queue_name}` - Main message queue
+- `retry:{queue_name}` - Retry queue with scheduled messages
+- `dead_letter:{queue_name}` - Dead letter queue for failed messages
+
+### Message Format
+
+Messages are stored in JSON format with the following structure:
+
+```json
+{
+    "id": "uuid",
+    "queue": "queue_name",
+    "payload": {
+        // Your message data
+    },
+    "created_at": "2023-08-26T12:00:00Z",
+    "retry_count": 0,
+    "max_retries": 3,
+    "next_retry_at": null,
+    "error": null
+}
+```
+
+## Development
+
+1. Create a virtual environment:
+```bash
+python -m venv venv
+source venv/bin/activate  # Linux/MacOS
+# or
+.\\venv\\Scripts\\activate  # Windows
+```
+
+2. Install development dependencies:
+```bash
+pip install -e ".[dev]"
+```
+
+3. Run tests:
+```bash
+pytest
+```
+
+## License
+
+MIT License. See LICENSE file for details.

async_framework/__init__.py

@@ -0,0 +1,7 @@
+from .broker import MessageBroker
+from .consumer import MessageConsumer
+from .config import BrokerConfig, RedisConfig, RetryConfig
+from .message import Message
+
+__version__ = "0.1.0"
+__all__ = ["MessageBroker", "MessageConsumer", "BrokerConfig", "RedisConfig", "RetryConfig", "Message"]

async_framework/broker.py

@@ -0,0 +1,104 @@
+import redis
+from typing import Dict, Any
+from datetime import datetime, timedelta
+import structlog
+from prometheus_client import Counter, start_http_server
+
+from .config import BrokerConfig
+from .message import Message
+
+logger = structlog.get_logger()
+
+messages_published = Counter(
+    "messages_published_total",
+    "Total number of messages published",
+    ["queue"]
+)
+
+class MessageBroker:
+    def __init__(self, config: BrokerConfig):
+        self.config = config
+        logger.info("Creating Redis connection pool", 
+                    host=config.redis.host,
+                    port=config.redis.port,
+                    ssl=config.redis.ssl)
+                    
+        # Base connection parameters
+        connection_params = {
+            "host": config.redis.host,
+            "port": config.redis.port,
+            "db": config.redis.db,
+            "password": config.redis.password,
+            "decode_responses": True,
+            "max_connections": config.redis.connection_pool_size
+        }
+
+        # Add SSL configuration if enabled
+        if config.redis.ssl:
+            connection_params.update({
+                "ssl": True,
+                "ssl_cert_reqs": None,
+                "ssl_ca_certs": None
+            })
+
+        connection_pool = redis.ConnectionPool(**connection_params)
+        self._redis = redis.Redis(connection_pool=connection_pool)
+        # Start Prometheus metrics server
+        start_http_server(config.metrics_port)
+        logger.info("Message broker initialized", config=config.dict())
+
+    def publish(self, queue: str, payload: Dict[str, Any], max_retries: int = None) -> Message:
+        """Publish a message to a queue."""
+        message = Message(
+            queue=queue,
+            payload=payload,
+            max_retries=max_retries or self.config.retry.max_retries
+        )
+        
+        try:
+            self._redis.lpush(f"queue:{queue}", message.to_json())
+            messages_published.labels(queue=queue).inc()
+            logger.info("Message published", message_id=message.id, queue=queue)
+            return message
+        except redis.RedisError as e:
+            logger.error("Failed to publish message", error=str(e), queue=queue)
+            raise
+
+    def retry_message(self, message: Message) -> None:
+        """Move a message to the retry queue."""
+        if message.retry_count >= message.max_retries:
+            self._move_to_dead_letter(message)
+            return
+
+        message.retry_count += 1
+        delay = min(
+            self.config.retry.initial_delay * (self.config.retry.backoff_factor ** (message.retry_count - 1)),
+            self.config.retry.max_delay
+        )
+        message.next_retry_at = datetime.utcnow() + timedelta(seconds=delay)
+        
+        try:
+            self._redis.lpush(f"retry:{message.queue}", message.to_json())
+            logger.info(
+                "Message moved to retry queue",
+                message_id=message.id,
+                queue=message.queue,
+                retry_count=message.retry_count
+            )
+        except redis.RedisError as e:
+            logger.error("Failed to move message to retry queue", error=str(e))
+            raise
+
+    def _move_to_dead_letter(self, message: Message) -> None:
+        """Move a message to the dead letter queue."""
+        try:
+            self._redis.lpush(f"dead_letter:{message.queue}", message.to_json())
+            logger.warning(
+                "Message moved to dead letter queue",
+                message_id=message.id,
+                queue=message.queue,
+                retry_count=message.retry_count
+            )
+        except redis.RedisError as e:
+            logger.error("Failed to move message to dead letter queue", error=str(e))
+            raise

async_framework/config.py

@@ -0,0 +1,24 @@
+from typing import Optional
+from pydantic import BaseModel, Field
+
+class RedisConfig(BaseModel):
+    host: str = Field(default="localhost")
+    port: int = Field(default=6379)
+    db: int = Field(default=0)
+    password: Optional[str] = Field(default=None)
+    ssl: bool = Field(default=False)
+    connection_pool_size: int = Field(default=10)
+
+class RetryConfig(BaseModel):
+    max_retries: int = Field(default=3)
+    initial_delay: float = Field(default=1.0)  # seconds
+    max_delay: float = Field(default=60.0)  # seconds
+    backoff_factor: float = Field(default=2.0)
+
+class BrokerConfig(BaseModel):
+    redis: RedisConfig = Field(default_factory=RedisConfig)
+    retry: RetryConfig = Field(default_factory=RetryConfig)
+    num_workers: int = Field(default=1)
+    batch_size: int = Field(default=10)
+    polling_interval: float = Field(default=1.0)  # seconds
+    metrics_port: int = Field(default=8000)

async_framework/consumer.py

@@ -0,0 +1,202 @@
+import redis
+from typing import Callable, List, Optional
+import time
+from concurrent.futures import ThreadPoolExecutor
+import structlog
+from prometheus_client import Counter, Histogram
+
+from .config import BrokerConfig
+from .message import Message
+
+logger = structlog.get_logger()
+
+messages_processed = Counter(
+    "messages_processed_total",
+    "Total number of messages processed",
+    ["queue", "status"]
+)
+
+processing_time = Histogram(
+    "message_processing_seconds",
+    "Time spent processing messages",
+    ["queue"]
+)
+
+class MessageConsumer:
+    def __init__(self, config: BrokerConfig, queue: str, handler: Callable[[Message], None]):
+        self.config = config
+        self.queue = queue
+        self.handler = handler
+        logger.info("Creating Redis connection pool", 
+                    host=config.redis.host,
+                    port=config.redis.port,
+                    ssl=config.redis.ssl)
+        
+        connection_params = {
+            "host": config.redis.host,
+            "port": config.redis.port,
+            "db": config.redis.db,
+            "password": config.redis.password,
+            "decode_responses": True,
+            "max_connections": config.redis.connection_pool_size
+        }
+
+        if config.redis.ssl:
+            connection_params.update({
+                "ssl": True,
+                "ssl_cert_reqs": None,
+                "ssl_ca_certs": None
+            })
+
+        connection_pool = redis.ConnectionPool(**connection_params)
+        self._redis = redis.Redis(connection_pool=connection_pool)
+        self._executor = ThreadPoolExecutor(max_workers=config.num_workers)
+        self._running = False
+        logger.info("Message consumer initialized", queue=queue, config=config.dict())
+
+    def start(self) -> None:
+        """Start consuming messages."""
+        self._running = True
+        self._executor.submit(self._process_retry_queue)
+        logger.info("self.config.num_workers")
+        for _ in range(self.config.num_workers):
+            logger.info("-----------------")
+            self._executor.submit(self._consume)
+        logger.info("Consumer started", queue=self.queue)
+
+    def stop(self) -> None:
+        """Stop consuming messages."""
+        self._running = False
+        self._executor.shutdown(wait=True)
+        logger.info("Consumer stopped", queue=self.queue)
+
+    def _consume(self) -> None:
+        """Consume messages from the queue."""
+        logger.info("Consumer thread started", queue=self.queue)
+        while self._running:
+            try:
+                messages = self._batch_pop_messages()
+                if messages:
+                    logger.info("Received messages", queue=self.queue, count=len(messages))
+                    for message_data in messages:
+                        self._process_message(Message.from_json(message_data))
+                else:
+                    # Small sleep to prevent CPU spinning when queue is empty
+                    time.sleep(0.1)
+            except Exception as e:
+                logger.error("Error in consumer loop", error=str(e), queue=self.queue)
+                time.sleep(1)
+
+    def _batch_pop_messages(self) -> List[str]:
+        """Pop a batch of messages from the queue."""
+        messages = []
+        try:
+            # Using brpop instead of rpop for blocking operation
+            result = self._redis.brpop([f"queue:{self.queue}"], timeout=1)
+            if result:
+                messages.append(result[1])  # brpop returns (key, value) tuple
+                
+                # Try to get more messages up to batch size
+                for _ in range(self.config.batch_size - 1):
+                    msg = self._redis.rpop(f"queue:{self.queue}")
+                    if msg:
+                        messages.append(msg)
+                    else:
+                        break
+                        
+            logger.debug("Batch pop result", 
+                        queue=self.queue, 
+                        messages_count=len(messages))
+            return messages
+        except Exception as e:
+            logger.error("Error in batch pop", error=str(e), queue=self.queue)
+            return []
+
+    def _process_message(self, message: Message) -> None:
+        """Process a single message."""
+        with processing_time.labels(queue=self.queue).time():
+            try:
+                self.handler(message)
+                messages_processed.labels(
+                    queue=self.queue, status="success"
+                ).inc()
+                logger.info(
+                    "Message processed successfully",
+                    message_id=message.id,
+                    queue=self.queue
+                )
+            except Exception as e:
+                messages_processed.labels(
+                    queue=self.queue, status="error"
+                ).inc()
+                message.error = str(e)
+                self._handle_processing_error(message)
+
+    def _handle_processing_error(self, message: Message) -> None:
+        """Handle a message processing error."""
+        if message.retry_count < message.max_retries:
+            self._retry_message(message)
+        else:
+            self._move_to_dead_letter(message)
+
+    def _retry_message(self, message: Message) -> None:
+        """Move a message to the retry queue with exponential backoff."""
+        message.retry_count += 1
+        delay = min(
+            self.config.retry.initial_delay * 
+            (self.config.retry.backoff_factor ** (message.retry_count - 1)),
+            self.config.retry.max_delay
+        )
+        self._redis.zadd(
+            f"retry:{self.queue}",
+            {message.to_json(): time.time() + delay}
+        )
+        logger.info(
+            "Message scheduled for retry",
+            message_id=message.id,
+            queue=self.queue,
+            retry_count=message.retry_count,
+            delay=delay
+        )
+
+    def _process_retry_queue(self) -> None:
+        """Process messages in the retry queue."""
+        while self._running:
+            try:
+                # Get messages that are ready to be retried
+                messages = self._redis.zrangebyscore(
+                    f"retry:{self.queue}",
+                    "-inf",
+                    time.time(),
+                    start=0,
+                    num=self.config.batch_size
+                )
+                
+                if not messages:
+                    time.sleep(self.config.polling_interval)
+                    continue
+
+                # Remove the processed messages from the retry queue
+                pipeline = self._redis.pipeline()
+                for message_data in messages:
+                    #message = Message.from_json(message_data)
+                    pipeline.zrem(f"retry:{self.queue}", message_data)
+                    pipeline.lpush(f"queue:{self.queue}", message_data)
+                pipeline.execute()
+
+            except Exception as e:
+                logger.error("Error processing retry queue", error=str(e))
+                time.sleep(1)
+
+    def _move_to_dead_letter(self, message: Message) -> None:
+        """Move a message to the dead letter queue."""
+        try:
+            self._redis.lpush(f"dead_letter:{self.queue}", message.to_json())
+            logger.warning(
+                "Message moved to dead letter queue",
+                message_id=message.id,
+                queue=self.queue,
+                error=message.error
+            )
+        except redis.RedisError as e:
+            logger.error("Failed to move message to dead letter queue", error=str(e))

async_framework/message.py

@@ -0,0 +1,35 @@
+from datetime import datetime
+from typing import Any, Dict, Optional
+from pydantic import BaseModel, Field
+import json
+import uuid
+
+class Message(BaseModel):
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    queue: str
+    payload: Dict[str, Any]
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    retry_count: int = Field(default=0)
+    max_retries: int = Field(default=3)
+    next_retry_at: Optional[datetime] = Field(default=None)
+    error: Optional[str] = Field(default=None)
+
+    def to_json(self) -> str:
+        return json.dumps({
+            "id": self.id,
+            "queue": self.queue,
+            "payload": self.payload,
+            "created_at": self.created_at.isoformat(),
+            "retry_count": self.retry_count,
+            "max_retries": self.max_retries,
+            "next_retry_at": self.next_retry_at.isoformat() if self.next_retry_at else None,
+            "error": self.error
+        })
+
+    @classmethod
+    def from_json(cls, data: str) -> "Message":
+        json_data = json.loads(data)
+        json_data["created_at"] = datetime.fromisoformat(json_data["created_at"])
+        if json_data["next_retry_at"]:
+            json_data["next_retry_at"] = datetime.fromisoformat(json_data["next_retry_at"])
+        return cls(**json_data)

pyproject.toml

@@ -0,0 +1,20 @@
+[build-system]
+requires = ["setuptools>=45", "wheel>=0.36.0", "build>=0.7.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "async-framework"
+version = "0.1.0"
+description = "A Redis-based message queue framework with monitoring and retry capabilities"
+readme = "README.md"
+requires-python = ">=3.10"
+license = {text = "MIT"}
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+]
+
+[tool.setuptools]
+packages = ["async_framework"]

setup.py

@@ -0,0 +1,27 @@
+from setuptools import setup, find_packages
+
+setup(
+    name="async-framework",
+    version="0.1.0",
+    packages=find_packages(exclude=["tests*"]),
+    install_requires=[
+        "redis>=4.5.0",
+        "pydantic>=2.0.0",
+        "prometheus-client>=0.17.0",
+        "tenacity>=8.2.0",
+        "structlog>=23.1.0",
+    ],
+    python_requires=">=3.10",
+    author="Your Name",
+    author_email="your.email@example.com",
+    description="A Redis-based message queue framework with monitoring and retry capabilities",
+    long_description=open("README.md").read(),
+    long_description_content_type="text/markdown",
+    keywords="redis, queue, broker, consumer, async",
+    classifiers=[
+        "Development Status :: 3 - Alpha",
+        "Intended Audience :: Developers",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+    ],
+)