Metadata-Version: 2.1
Name: taskflows
Version: 0.1.5
Summary: Python task management, scheduling, alerts.
Author-email: Dan Kelleher <kelleherjdan@gmail.com>
License: MIT
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Description-Content-Type: text/markdown
Requires-Dist: sqlalchemy >=2.0.0
Requires-Dist: pydantic >=2.0.0
Requires-Dist: tqdm
Requires-Dist: click
Requires-Dist: python-dotenv
Requires-Dist: rich
Requires-Dist: dynamic-imports >=1.0.0
Requires-Dist: alert-msgs >=0.2.0
Requires-Dist: quicklogs >=1.1.0
Requires-Dist: func-timeout >=4.0.0
Provides-Extra: dev
Requires-Dist: black ; extra == 'dev'
Requires-Dist: pytest ; extra == 'dev'
Requires-Dist: psycopg >=3.0.0 ; extra == 'dev'


## 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..
```python
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
```python
from taskflows import ScheduledTask, OnCalendar

task = ScheduledTask(
    task_name="my-task",
    command="my-command",
    timer=OnCalendar("Sun 17:00 America/New_York"),
)
task.create()
```





