Crono is a programmatic time-based job scheduler that gives your application a sense of timing.
import crono
crono.request('POST', 'https://your.app/').after(hours=42)Install package:
pip install cronoRun the servers:
redis-server &
celery worker --app=crono.queue:queue --hostname=worker1@%h
celery beat --app=crono.queue:queueA trigger defines when a job will be executed. There are 4 types of triggers: after, on, every, and cron.
after
after specifies a countdown until the execution of a task. It will only occur once. It takes at least 1 keyword argument: hours, minutes, and/or seconds, of type . Those keywords arguments are cumulative (same for the every trigger). For example, hours=1, minutes=30 equals minutes=90.
crono.after(minutes=30).…on
on specifies the execution of a task at a specific date and time. It will only occur once. It takes 1 positional argument of type <datetime.datetime>.
import datetime
date = datetime.datetime(2019, 7, 4)
crono.on(date).…every
every specifies a frequency at which to execute a task. It will occur multiple times. It takes at least 1 keyword argument: hours, minutes, and/or seconds, of type . Those keywords arguments are cumulative (similarly to the after trigger). For example, hours=1, minutes=30 equals minutes=90.
crono.every(hours=1, minutes=30).…cron
cron uses an expression to specify the execution time. It will occur mutiple times. It takes exactly 1 positional argument of type <str>.
crono.cron('0 6 * * 2').…A task defines what a job will do. There are 4 types of tasks: log, request, message, and email.
log
log uses the standard logging Python library.
crono.log('DEBUG', '{text}', *args, **kwargs)request
request sends an HTTP request. It is powered by the Requests library.
crono.request('POST', '{url}', **kwargs).…message
message sends an SMS. It is powered by Twilio. To use it, you will have to specify twilio_account_sid and twilio_auth_id.
Not implemented, yet.
email sends an email. It is powered by Postmark. To use it, you will have to specify postmark_api_key and postmark_sender.
Not implemented, yet.
import crono
# Timer
crono.request('POST', '{url}').after(minutes=1)
# Datetime
crono.log('DEBUG', '{text}').on(<datetime>)
# Interval
crono.email(…).every(hours=1) # `email` task not implemented (yet)
# Cron
crono.message(…).cron('0 6 * * 2') # `message` task not implemented (yet)Crono comes with sensible default values that you can override:
# Required
CELERY_BROKER
# Optional
CELERY_RESULT_BACKEND
REDIS_MAX_CONNECTIONS (default: 20)
CELERY_BROKER_POOL_LIMIT (default: 0)
CELERY_TASK_IGNORE_RESULT (default: True)
CELERY_BEAT_MAX_LOOP_INTERVAL (default: 300)
CELERY_WORKER_MAX_TASKS_PER_CHILD (default: 100)
python -m pytest
Packaging
# Generating distribution archives
$ python setup.py sdist bdist_wheel
# Uploading the distribution archives
$ twine upload --skip-existing dist/*