Simple offline task queues. For Python.
"See you later, alligator."
Latest documentation at http://alligator.readthedocs.org/en/latest/.
- Python 3.6+
- (Optional)
redisfor the Redis backend - (Optional)
boto3>=1.12.0for the SQS backend
This example uses Django, but there's nothing Django-specific about Alligator.
I repeat, You can use it with any Python code that would benefit from background processing.
from alligator import Gator from django.contrib.auth.models import User from django.shortcuts import send_email # Make a Gator instance. # Under most circumstances, you would configure this in one place & # import that instance instead. gator = Gator('redis://localhost:6379/0') # The task itself. # Nothing special, just a plain *undecorated* function. def follow_email(followee_username, follower_username): followee = User.objects.get(username=followee_username) follower = User.objects.get(username=follower_username) subject = 'You got followed!' message = 'Hey {}, you just got followed by {}! Whoohoo!'.format( followee.username, follower.username ) send_email(subject, message, 'server@example.com', [followee.email]) # An simple, previously expensive view. @login_required def follow(request, username): # You'd import the task function above. if request.method == 'POST': # Schedule the task. # Use args & kwargs as normal. gator.task(follow_email, request.user.username, username) return redirect('...')
Rather than trying to do autodiscovery, fanout, etc., you control how your workers are configured & what they consume.
If your needs are simple, run the included latergator.py worker:
$ python latergator.py redis://localhost:6379/0
If you have more complex needs, you can create a new executable file (bin script, management command, whatever) & drop in the following code.
from alligator import Gator, Worker # Bonus points if you import that one pre-configured ``Gator`` instead. gator = Gator('redis://localhost:6379/0') # Consume & handle all tasks. worker = Worker(gator) worker.run_forever()
New BSD
Alligator has 95%+ test coverage & aims to be passing/stable at all times.
If you'd like to run the tests, clone the repo, then run:
$ virtualenv -p python3 env $ . env/bin/activate $ pip install -r requirements-tests.txt $ python setup.py develop $ pytest -s -v --cov=alligator --cov-report=html tests
The full test suite can be run via:
$ export ALLIGATOR_TESTS_INCLUDE_SQS=true $ ./tests/run_all.sh
This requires all backends/queues to be running, as well as valid AWS
credentials if ALLIGATOR_TESTS_INCLUDE_SQS=true is set.
- Because I have NIH-syndrome.
- Or because I longed for something simple (~375 loc).
- Or because I wanted something with tests (90%+ coverage) & docs.
- Or because I wanted pluggable backends.
- Or because testing some other queuing system was a pain.
- Or because I'm an idiot.
Post-1.0.0:
- Expand the supported backends
- Kafka?
- ActiveMQ support?
- RabbitMQ support?
- ???