Python task management, scheduling, alerts.
Project description
Task Management, Scheduling, and Alerting.
Admin commands are accessed via the taskflows
command line tool. See taskflows --help
for complete usage.
Setup
For task decorators:
pip install taskflows
For additional service/scheduling functionality:
sudo apt install libdbus-glib-1-dev
loginctl enable-linger
pip install taskflows[service]
Task execution metadata is stored in SQLite (default) or Postgresql. To use a personal database, set environment variable TASKFLOWS_DB_URL
to your database URL. If using Postgresql, TASKFLOWS_DB_SCHEMA may also be set to use a custom schema (default schema is taskflows).
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"),
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. Internally, alerts are sent using the alert-msgs package.
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 Services
Note: To use services, your system must have systemd (the init system on most modern Linux distributions)
Services run commands on a specified schedule. See Service for service configuration options.
To create the service(s), use the create
method (e.g. srv.create()
), or use the CLI create
command (e.g. taskflows create my_services.py
)
Examples
from taskflows import Calendar, Service
Run at specified calendar days/time.
see Calendar for more options.
srv = Service(
name="something",
command="docker start something",
schedule=Calendar("Mon-Sun 14:00 America/New_York"),
)
Run command once at half an hour from now.
run_time = datetime.now() + timedelta(minutes=30)
srv = Service(
name='write-message',
command="bash -c 'echo hello >> hello.txt'",
schedule=Calendar.from_datetime(run_time),
)
Run command after system boot, then again every 5 minutes after start of previous run. Skip run if CPU usage is over 80% for the last 5 minutes.
see Periodic and constraints for more options.
Service(
name="my-periodic-task",
command="docker start something",
schedule=Periodic(start_on="boot", period=60*5, relative_to="start"),
system_load_constraints=CPUPressure(max_percent=80, timespan="5min", silent=True)
)
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.7.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 2e17adb1cf8328bdaf148b8aafbcd6f7962a59ce94f01cdedcb2d58144b87e0c |
|
MD5 | 45eec7df285e5a2ab7c7b35e4eaf62e3 |
|
BLAKE2b-256 | 021a4a873768a829467af9ea30a40dbcbffae1efec15ed90f8d8e8fe9b821e3c |