installation.md¶

Installation¶

Requirements¶

Python 3.10 or higher
pip package manager

Install from PyPI¶

pip install flowtask

Install from Source¶

git clone https://github.com/phenobarbital/flowtask.git
cd flowtask
pip install -e .

Verify Installation¶

python -c "import flowtask; print(flowtask.__version__)"

documentation/docs/getting-started/quickstart.md¶

Quick Start¶

Basic Task Structure¶

Tasks in FlowTask are organized in a directory structure:

tasks/
├── programs/
│   ├── test/
│   │   ├── tasks/
│   │   │   └── example.yaml

Your First Task¶

Create a simple task file tasks/programs/test/tasks/example.yaml:

name: Hello World
description: A simple example task
steps:
  - Echo:
      message: "Hello, FlowTask!"

Running Tasks¶

From CLI¶

task --program=test --task=example

With Environment¶

ENV=dev task --program=test --task=example

Programmatically¶

from flowtask import Task
import asyncio

task = Task(program='test', task='example')
results = asyncio.run(task.run())
print(results)

Next Steps¶

Explore Components to see available processing units
Learn about Configuration options
Check out Examples for real-world use cases

documentation/docs/getting-started/configuration.md¶

Configuration¶

Task Storage¶

FlowTask supports multiple storage backends for tasks:

Filesystem (Default)¶

# Tasks stored in local filesystem
task_storage = DirectoryTaskStorage("/path/to/tasks")

Database¶

# Tasks stored in database
task_storage = DatabaseTaskStorage(connection_string)

S3 Bucket¶

# Tasks stored in S3
task_storage = S3TaskStorage(bucket_name, region)

Environment Variables¶

FlowTask uses environment files for configuration:

# .env
DATABASE_URL=postgresql://user:pass@localhost/db
S3_BUCKET=my-task-bucket
REDIS_URL=redis://localhost:6379

Task Configuration¶

Basic Structure¶

name: Task Name
description: Task description
environment: production  # optional
timeout: 3600  # optional, in seconds
steps:
  - ComponentName:
      parameter1: value1
      parameter2: value2

Advanced Features¶

name: Advanced Task
description: Task with advanced features
steps:
  - InputComponent:
      # Component configuration
  - ProcessingComponent:
      # Can reference previous step output
      input_data: "{{ previous.output }}"
  - OutputComponent:
      # Final processing

HTTP API Configuration¶

When running as an HTTP service:

# app.py
from flowtask.services.tasks import TaskManager

app.router.add_view('/api/v2/tasks/{program}/{task_id}', TaskManager)

Available endpoints: - GET /api/v2/tasks - List all tasks - GET /api/v2/tasks/{program} - List tasks in program - POST /api/v2/task/{program}/{task_id} - Execute task - GET /api/v2/tasks/{program}/{task_id} - Get task details