enqueuefy 0.0.1a1

Async task queue library for Python — Redis and AWS SQS/Lambda backends

enqueuefy

Async task queue library for Python with Redis and AWS SQS/Lambda backends.

Installation

pip install enqueuefy

Quick start — Redis

server.py (worker process)

from enqueuefy import AIORedisConsumer

consumer = AIORedisConsumer(
    broker_url="redis://localhost:6379",
    modules=["myapp.tasks"],
)

queue = consumer.queue("default")

if __name__ == "__main__":
    consumer.run()

tasks.py (define tasks)

from myapp.server import queue

@queue.task(retries=3, timeout=30)
async def send_email(to: str, subject: str):
    ...

@send_email.on_failure
async def on_send_failure(to: str, subject: str):
    print(f"Failed to send email to {to}")

@send_email.on_timeout
async def on_send_timeout(to: str, subject: str):
    print(f"Email to {to} timed out")

producer.py (enqueue tasks)

from myapp.tasks import send_email

await send_email.enqueue("[e-mail]", subject="Hello")

Run the worker

python server.py --broker-url redis://localhost:6379 --queues default --max-concurrency 10

Quick start — AWS SQS / Lambda

server.py

from enqueuefy import AIOSQSLambdaConsumer

consumer = AIOSQSLambdaConsumer(
    region_name="us-east-1",
    aws_access_key_id="...",
    aws_secret_access_key="...",
    modules=["myapp.tasks"],
)

queue = consumer.queue("https://sqs.us-east-1.amazonaws.com/123456789/my-queue")

if __name__ == "__main__":
    consumer.run()  # generates Lambda handler files

tasks.py

from myapp.server import queue

@queue.task()
async def process_order(order_id: str):
    ...

Running server.py generates ready-to-deploy AWS Lambda handler files in sqs_lambda_handlers/.

CLI options (Redis worker)

Flag	Default	Description
--broker-url	—	Redis connection URL
--queues	default	Comma-separated queue names
--max-concurrency	10	Max concurrent tasks

Features

• Async-first, built on asyncio
• Redis backend with at-least-once delivery and startup recovery of in-flight messages
• Per-task retries, timeout, on_failure, and on_timeout hooks
• SQS/Lambda backend that auto-generates Lambda handler files
• Distributed startup-requeue lock — safe for multi-pod Kubernetes deployments

Development

make test        # run tests
make coverage    # test with coverage report
make format      # black + isort
make build       # build distribution packages
make publish     # upload to PyPI via twine
make publish-test  # upload to TestPyPI

License

MIT