import asyncio from kew import TaskQueueManager, QueueConfig, QueuePriority async def process_order(order_id: str): # Simulate order processing await asyncio.sleep(1) return f"Order {order_id} processed" async def main(): # Initialize queue manager manager = TaskQueueManager(redis_url="redis://localhost:6379") await manager.initialize() # Create processing queue await manager.create_queue(QueueConfig( name="orders", max_workers=4, # Only 4 concurrent tasks max_size=1000 )) # Submit some tasks tasks = [] for i in range(10): task = await manager.submit_task( task_id=f"order-{i}", queue_name="orders", task_type="process_order", task_func=process_order, priority=QueuePriority.MEDIUM, order_id=str(i) ) tasks.append(task) # Check results # Small delay to allow tasks to complete in this simple example await asyncio.sleep(1.2) for task in tasks: status = await manager.get_task_status(task.task_id) print(f"{task.task_id}: {status.result}") if __name__ == "__main__": asyncio.run(main())

# Strictly enforce 4 concurrent tasks max await manager.create_queue(QueueConfig( name="api_calls", max_workers=4 # Guaranteed not to exceed ))

# High priority queue for urgent tasks await manager.create_queue(QueueConfig( name="urgent", priority=QueuePriority.HIGH )) # Lower priority for batch processing await manager.create_queue(QueueConfig( name="batch", priority=QueuePriority.LOW ))

await manager.create_queue(QueueConfig( name="flaky_api", max_workers=4, max_retries=3, # Retry up to 3 times on failure retry_delay=1.0, # Base delay of 1 second (doubles each retry) )) # Tasks that fail will be re-queued automatically: # Attempt 1: immediate # Attempt 2: +1s delay # Attempt 3: +2s delay # Attempt 4: +4s delay (or fail permanently)

from datetime import datetime, timedelta # Defer by a duration await manager.submit_task( task_id="send-reminder", queue_name="emails", task_type="reminder", task_func=send_reminder, priority=QueuePriority.MEDIUM, _defer_by=300.0, # Execute 5 minutes from now user_id="abc123", ) # Defer until a specific time await manager.submit_task( task_id="morning-report", queue_name="reports", task_type="report", task_func=generate_report, priority=QueuePriority.LOW, _defer_until=datetime(2025, 1, 15, 9, 0, 0), # Run at 9 AM )

async def on_start(task_info): print(f"Task {task_info.task_id} started") async def on_complete(task_info): await metrics.record("task.completed", task_info.task_id) async def on_fail(task_info, error): await alert_channel.send(f"Task {task_info.task_id} failed: {error}") manager = TaskQueueManager( redis_url="redis://localhost:6379", on_task_start=on_start, on_task_complete=on_complete, on_task_fail=on_fail, )

await manager.create_queue(QueueConfig( name="external_api", max_workers=4, max_circuit_breaker_failures=5, # Open after 5 consecutive failures circuit_breaker_reset_timeout=30, # Auto-close after 30 seconds ))

from kew.exceptions import QueueProcessorError await manager.create_queue(QueueConfig( name="bounded_queue", max_workers=2, max_size=100, # Reject submissions beyond 100 queued tasks )) try: await manager.submit_task(...) except QueueProcessorError: # Queue is full - apply backpressure to caller return {"status": "busy", "retry_after": 5}

tasks = [ { "task_id": f"order-{i}", "task_type": "process", "task_func": process_order, "priority": QueuePriority.MEDIUM, "kwargs": {"order_id": i}, } for i in range(1000) ] # Single call, batched internally in chunks of 50 results = await manager.submit_tasks("orders", tasks) # ~33,000 tasks/sec — 12x faster than sequential submit_task()

# Check task status status = await manager.get_task_status("task-123") print(f"Status: {status.status}") print(f"Result: {status.result}") print(f"Error: {status.error}") print(f"Retries: {status.retry_count}") # Get all currently running tasks ongoing = await manager.get_ongoing_tasks() # Monitor queue health queue_status = await manager.get_queue_status("api_calls") print(f"Active Tasks: {queue_status['current_workers']}") print(f"Circuit Breaker: {queue_status['circuit_breaker_status']}")

from fastapi import FastAPI from kew import TaskQueueManager, QueueConfig, QueuePriority app = FastAPI() manager = TaskQueueManager() @app.on_event("startup") async def startup(): await manager.initialize() await manager.create_queue(QueueConfig( name="emails", max_workers=2, max_retries=3, # Retry failed email sends retry_delay=5.0, # 5s base backoff )) @app.post("/signup") async def signup(email: str): # Handle signup immediately user = await create_user(email) # Queue welcome email for background processing await manager.submit_task( task_id=f"welcome-{user.id}", queue_name="emails", task_type="send_welcome_email", task_func=send_welcome_email, priority=QueuePriority.MEDIUM, user_id=user.id ) return {"status": "success"}

manager = TaskQueueManager( redis_url="redis://username:password@hostname:6379/0", cleanup_on_start=True # Optional: clean stale tasks )

try: await manager.submit_task(...) except TaskAlreadyExistsError: # Handle duplicate task except QueueProcessorError as e: # Handle processing error or queue full print(f"Task failed: {e}")