celery-batches

from celery import Celery from celery_batches import Batches import os # Configure Celery app (replace with your actual broker/backend) app = Celery('my_app', broker=os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0'), backend=os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1')) @app.task(base=Batches, flush_every=10, flush_interval=5) # Flush after 10 messages or 5 seconds def process_batch_task(requests): print(f"Processing a batch of {len(requests)} requests.") for request in requests: data = request.kwargs.get('data') task_id = request.id print(f" - Task ID: {task_id}, Data: {data}") # If you need to return results for individual tasks, do it here: # app.backend.mark_as_done(task_id, f"Processed: {data}", request=request) if __name__ == '__main__': # Example usage: send individual tasks print("Sending 15 tasks...") for i in range(15): process_batch_task.delay(data=f'item_{i}') print("Waiting for tasks to be processed by a Celery worker...") # To run a worker: celery -A your_module_name worker -l info --pool=solo --prefetch-multiplier=0 # Note: --pool=solo is for simple testing. For production, use a proper pool and ensure prefetch-multiplier=0.