Skip to main content

Programmatic time-based job scheduler

Project description

🔮 Crono

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)

Read more

Install

Install package:

pip install crono

Run the servers:

redis-server &
celery worker --app=crono.queue:queue --hostname=worker1@%h
celery beat --app=crono.queue:queue

Usage

Triggers

A 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 <int>. 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 <int>. 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').

Tasks

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

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.

Examples

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)

Configuration

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)

Test

python -m pytest

Development

Packaging

# Generating distribution archives
$ python setup.py sdist bdist_wheel

# Uploading the distribution archives
$ twine upload --skip-existing dist/*

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for crono, version 0.1.4
Filename, size File type Python version Upload date Hashes
Filename, size crono-0.1.4-py3-none-any.whl (7.0 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size crono-0.1.4.tar.gz (4.9 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page