Python task management, scheduling, alerts.
Project description
Python task management, scheduling, alerts.
Admin commands are accessed via the tasks
command line tool. See tasks --help
for complete usage.
Setup
pip install taskflows
Task execution metadata is stored in Postgresql. Set environment variable TASKFLOWS_DB
to the URL of the Postgresql instance you would like to use.
Create Tasks
Turn any function (optionally async) into a task that logs metadata to the database and sends alerts, allows retries, etc..
alerts=[
Alerts(
send_to=[
Slack(
bot_token=os.getenv("SLACK_BOT_TOKEN"),
app_token=os.getenv("SLACK_APP_TOKEN"),
channel="critical_alerts"
),
Email(
addr="sender@gmail.com",
password=os.getenv("EMAIL_PWD"),
receiver_addr=["someone@gmail.com", "someone@yahoo.com"]
)
],
send_on=["start", "error", "finish"]
)
]
@task(
name='some-task',
required=True,
retries=1,
timeout=30,
alerts=alerts
)
async def hello():
print("Hi.")
Review Task Status/Results
Tasks can send alerts via Slack and/or Email, as shown in the above example. Task start/finish times, status, retry count, return values can be found in the task_runs
table. Any errors that occurred during the execution of a task can be found in the task_errors
table.
Create a scheduled Task
from taskflows import ScheduledTask, OnCalendar
task = ScheduledTask(
task_name="my-task",
command="my-command",
timer=OnCalendar("Sun 17:00 America/New_York"),
)
task.create()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Hashes for taskflows-0.1.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | d5e81d907143e531a5415100e549ee4ef9e1ab3d2aa789d096b28c754be204eb |
|
MD5 | 99e979e52b5e981ac5fcb2054786877c |
|
BLAKE2b-256 | 9d69beb0e0fe315a9a53f8335b425d98b5e8d19852113faa3b1a9ee41545a221 |