pyworkflow-framework 0.1.2

A lightweight, developer-friendly Python workflow automation framework.

PyWorkflow

A lightweight, developer-friendly Python workflow automation framework — think Airflow/Prefect/Dagster, but simple enough to learn in five minutes.

Define workflows as plain Python. Run them sequentially or in parallel. Get automatic retries, dependency resolution, scheduling, persistent history, a CLI, and a plugin system — with zero required infrastructure (no server, no database, no message broker).

from pyworkflow import Workflow, Task

workflow = Workflow("Data Processing")

workflow.add_task(Task("Download Data", download_function))
workflow.add_task(Task("Clean Data", clean_function, depends_on=["Download Data"]))
workflow.add_task(Task("Analyze Data", analyze_function, depends_on=["Clean Data"]))

report = workflow.run()
print(report.success, report.results)

Status

v0.1.0 — first release. The core engine (workflows, tasks, dependencies, retries, parallel execution, scheduling, storage, CLI) is stable and fully tested. Visualization and plugins are usable but considered early; see Roadmap.

---

Installation

pip install pyworkflow-framework

Optional extras:

pip install "pyworkflow-framework[viz]"   # graphviz/networkx-powered workflow.visualize()
pip install "pyworkflow-framework[dev]"   # pytest + pytest-cov for running the test suite

From source:

git clone https://github.com/Rapheal-Kwabena/pyworkflow.git
cd pyworkflow
pip install -e ".[dev]"

---

Quick Start

1. Define tasks as ordinary Python functions

def download_data():
    return {"rows": 1000}

def clean_data(context):
    # `context` holds the output of every completed task, keyed by task name
    raw = context["Download"]
    return {"rows": raw["rows"] - 10}  # drop 10 bad rows

def analyze_data(context):
    return f"Analyzed {context['Clean']['rows']} rows"

A task function may optionally accept a context: dict parameter — PyWorkflow detects this via introspection and injects it automatically. Tasks that don't need upstream data can omit it entirely.

2. Build and run a workflow

from pyworkflow import Workflow, Task

workflow = Workflow("Data Processing")
workflow.add_task(Task("Download", download_data))
workflow.add_task(Task("Clean", clean_data, depends_on=["Download"]))
workflow.add_task(Task("Analyze", analyze_data, depends_on=["Clean"]))

report = workflow.run()

if report.success:
    print("All done:", report.results)
else:
    print("Failed tasks:", report.failed_tasks)

3. Or build it fluently with .then(...)

workflow = Workflow("Data Processing")
(
    workflow
    .add_task(Task("Download", download_data))
    .then(Task("Clean", clean_data))
    .then(Task("Analyze", analyze_data))
)
workflow.run()

---

Examples

Parallel execution

Independent tasks (no shared dependency chain) run concurrently with parallel=True:

workflow = Workflow("Fan-out", max_workers=4)
workflow.add_task(Task("Fetch US", fetch_us))
workflow.add_task(Task("Fetch EU", fetch_eu))
workflow.add_task(Task("Fetch APAC", fetch_apac))
workflow.add_task(Task("Merge", merge_regions, depends_on=["Fetch US", "Fetch EU", "Fetch APAC"]))

workflow.run(parallel=True)  # the three Fetch tasks run concurrently, then Merge

Retries and failure handling

workflow.add_task(
    Task(
        "Send Email",
        send_email,
        retries=3,
        retry_delay=5,          # seconds between attempts
        continue_on_failure=True,  # don't halt the whole workflow if this fails
        on_failure=lambda task, exc: alert_ops_team(task.name, exc),
    )
)

report = workflow.run()
if not report.success:
    workflow.retry_failed_tasks()  # re-run only what failed

Conditional / branching tasks

workflow.add_task(Task("Check Inventory", check_inventory))
workflow.add_task(
    Task(
        "Reorder Stock",
        reorder_stock,
        depends_on=["Check Inventory"],
        condition=lambda context: context["Check Inventory"]["low_stock"] is True,
    )
)

Chaining two workflows

ingest = Workflow("Ingest").add_task(Task("Pull", pull_data))
report_wf = Workflow("Report").add_task(Task("Build Report", build_report))

combined = ingest.chain(report_wf)  # Report's tasks depend on Ingest's terminal tasks
combined.run()

Scheduling

workflow.schedule("now")                          # run immediately
workflow.schedule("delay", delay=30)               # run once, in 30 seconds
workflow.schedule("interval", every=15)             # run every 15 minutes
workflow.schedule("cron", cron="0 8 * * 1-5")        # 8am on weekdays
workflow.schedule("daily", time="08:00")             # shorthand for daily cron

Persisting workflow history

from pyworkflow.storage import JSONStorage, SQLiteStorage

storage = JSONStorage()  # defaults to ~/.pyworkflow/
storage.save_workflow(workflow.to_dict())
storage.save_run(workflow.name, {"success": report.success, "results": report.results})

print(storage.list_workflows())
print(storage.get_history(workflow.name))

# Or, if you'd rather query with SQL:
sql_storage = SQLiteStorage()  # defaults to ~/.pyworkflow/pyworkflow.db

Visualization

workflow.visualize()  # renders a PNG dependency graph if graphviz is installed,
                       # otherwise prints a readable text dependency map

Plugins

Plugins turn common integrations into reusable Task factories:

from pyworkflow.plugins import EmailPlugin, APIPlugin

email = EmailPlugin()
email.setup(host="smtp.example.com", username="[e-mail]", password="...")

api = APIPlugin()
api.setup(base_url="https://api.example.com", headers={"Authorization": "Bearer ..."})

workflow.add_task(email.make_task("Notify", to="[e-mail]", subject="Done", body="Workflow finished"))
workflow.add_task(api.make_task("Fetch Users", path="/users", method="GET"))

Write your own by subclassing pyworkflow.plugins.Plugin.

AI-powered steps (optional)

PyWorkflow doesn't hard-depend on any particular LLM SDK. Wire up whichever one you use via a plain call_fn(prompt) -> str:

from pyworkflow.plugins import AIPlugin
import anthropic

client = anthropic.Anthropic()

def call_claude(prompt: str) -> str:
    msg = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

ai = AIPlugin()
ai.setup(call_fn=call_claude)

workflow.add_task(ai.make_task("Summarize Feedback", prompt="Summarize this customer feedback."))
workflow.add_task(
    ai.make_decision_task("Route Ticket", question="Which team should handle this?",
                           choices=["billing", "engineering", "support"])
)

---

Command Line Interface

pyworkflow create myworkflow      # scaffold myworkflow.py with a starter workflow
pyworkflow run myworkflow.py      # run it (add --parallel to run independent tasks concurrently)
pyworkflow status "My Workflow"   # show last known state from local storage
pyworkflow history "My Workflow"  # show recent run history
pyworkflow logs "My Workflow"     # show failures/skips from the latest run
pyworkflow list                   # list all workflows in local storage

Since workflows are Python code (functions aren't serializable to JSON), the CLI operates on workflow definition files — a .py file exposing a module-level workflow = Workflow(...) variable, exactly like the file pyworkflow create scaffolds for you.

---

API Documentation

Workflow

Method	Description
add_task(task)	Add a Task. Returns self for chaining.
add_tasks(tasks)	Add multiple tasks at once.
then(task)	Add a task that depends on the most recently added one.
chain(other_workflow)	Combine two workflows; returns a new Workflow.
run(parallel=False)	Execute the workflow. Returns an ExecutionReport.
retry_failed_tasks(parallel=False)	Re-run only tasks left in FAILED state.
cancel() / pause() / resume()	Lifecycle control (in-process).
reset()	Reset workflow and all tasks to their initial state.
schedule(mode, ...)	Schedule via the shared background Scheduler.
visualize()	Render/print the dependency graph.
summary() / to_dict()	Structured status snapshots.

Task

Key constructor arguments: name, function, description, args, kwargs, retries, retry_delay, timeout, depends_on, condition, on_failure, continue_on_failure. See docstrings in pyworkflow/core/task.py for full details.

States

WorkflowState: CREATED, RUNNING, COMPLETED, FAILED, CANCELLED, PAUSED
TaskState:     PENDING, RUNNING, COMPLETED, FAILED, SKIPPED, RETRYING, CANCELLED

ExecutionReport

report.success        # bool
report.results        # dict[task_name, output]
report.failed_tasks    # list[str]
report.skipped_tasks   # list[str]
report.error           # first exception encountered, if any

---

Architecture

pyworkflow/
├── core/
│   ├── task.py          # Task, TaskState, TaskResult — a single unit of work
│   ├── workflow.py       # Workflow, WorkflowState — composition & orchestration API
│   └── engine.py         # Engine — topological execution (sequential/parallel)
├── scheduler/
│   ├── scheduler.py       # background-thread scheduler (now/delay/interval/cron)
│   └── cron.py             # dependency-free 5-field cron parser/matcher
├── storage/
│   ├── base.py              # StorageBackend interface
│   ├── json_storage.py       # local JSON files (default, zero setup)
│   └── sqlite_storage.py      # SQLite backend for queryable history
├── plugins/
│   ├── base.py                 # Plugin base class + PluginRegistry
│   ├── email.py, database.py, api.py, ai.py   # built-in Task factories
├── visualization/
│   └── graph.py                 # graphviz-based rendering with text fallback
├── cli/
│   └── main.py                    # `pyworkflow` command line entry point
└── exceptions/
    └── __init__.py                  # PyWorkflowError hierarchy

Design principles:

• Core has zero required third-party dependencies beyond click (for the CLI). Everything else (graphviz, DB drivers, LLM SDKs) is optional and only imported when you use that feature.
• The dependency graph is the source of truth. Engine computes execution order via a topological sort (Kahn's algorithm), so sequential and parallel execution share identical semantics — parallelism only changes how independent tasks run, never what runs.
• Context is immutable per task. Each task receives a snapshot dict of prior outputs; tasks can't accidentally mutate what other tasks see mid-run.
• Storage is pluggable. JSONStorage and SQLiteStorage both implement the same StorageBackend interface — swap one for the other without touching workflow code.

---

Testing

pip install -e ".[dev]"
pytest

The test suite covers workflow creation/composition, sequential and parallel task execution, dependency validation (missing deps, cycles), retries and failure callbacks, the scheduler and cron matcher, both storage backends, and the CLI.

---

Roadmap

Planned for future releases, with the architecture already designed to support them:

• Web dashboard for real-time monitoring
• Distributed / cloud workers (beyond in-process threads)
• Kubernetes-native workflow execution
• Real-time execution event streaming
• A workflow marketplace / template library
• An AI-assisted workflow builder

Contributing

1. Fork the repo and create a feature branch.
2. Add tests for any new behavior (tests/).
3. Run pytest and make sure everything passes.
4. Follow PEP 8 and keep type hints on public functions.
5. Open a pull request describing the change and why it's needed.

Bug reports and feature requests are welcome via GitHub Issues.

License

MIT — see LICENSE.