A distributed and durable workflow engine for Python.
Built by Ezequiel Ranieri — Backend & Security Engineer specialized in Distributed Systems
pip install flowcore-engineOr with uv:
uv add flowcore-engineI built Flowcore to solve a real problem I kept encountering: complex business processes that needed to survive failures, resume from where they stopped, and scale across workers. Flowcore is a lightweight yet powerful workflow execution engine designed to be durable and distributed. It allows defining complex business processes using an elegant DSL based on Python decorators, ensuring reliable step execution with automatic retries and state persistence at every transition.
from flowcore.domain.dsl.primitives import task, workflow
from flowcore.domain.dsl.models import Step
@task(name="validate_order", max_retries=3)
def validate_order(ctx: dict):
return {"valid": True}
@workflow(name="order_process", version="1.0.0")
class OrderWorkflow:
steps = [
Step(name="validate", task_name="validate_order", next_steps=["pay"]),
Step(name="pay", task_name="process_payment")
]Flowcore allows you to define complex, multi-step workflows in Python using a declarative DSL and execute them in a distributed environment with guaranteed persistence and resilience.
Flowcore includes real-world examples that demonstrate its core features:
-
Payment Pipeline with Saga: Automatic compensation when payment fails after inventory reservation. Run it standalone with
python examples/payment_with_saga.py. -
Parallel User Onboarding: Fan-out with join/barrier — account setup runs in parallel, activation waits for all branches. Run it standalone with
python examples/parallel_onboarding.py. -
Versioned Deployment: Two workflow versions running simultaneously without interference. Run it standalone with
python examples/versioned_deployment.py.
Each script runs immediately using the in-memory WorkflowEngine — no need to start the full stack. For example, running the saga script produces:
--- Running Payment Workflow with amount: $1500 ---
[Step] Validating payment amount...
[Saga Step] Reserving $1500 from user account...
[Saga Step] Reserving item SKU-123 in inventory...
[Step] Processing actual payment of $1500...
[!] Payment gateway rejected: High amount detected.
--- Workflow Failed: Payment gateway error: Transaction limit exceeded ---
Starting Saga Compensation Sequence...
[Compensation] Releasing inventory reservation...
[Compensation] Releasing reserved funds...
--- All compensations completed. System is consistent. ---
- DSL Reference
- Configuration
- How-to: First Workflow
- How-to: Sagas
- How-to: Multi-tenancy
- How-to: Observability
- Production Deployment Guide
# Clone the repository
git clone https://github.com/ezequielranieri/flowcore.git
cd flowcore
# Start the MVP stack
make up
# Apply migrations
make migrateTo trigger a real workflow execution, send a request to the API:
# Using curl
curl -X POST http://localhost:8000/workflows/order_process -H "Content-Type: application/json" -d '{}'
# Or using PowerShell/Invoke-RestMethod
Invoke-RestMethod -Uri http://localhost:8000/workflows/order_process -Method Post -ContentType "application/json" -Body '{}'
You can visualize and interact with the API endpoints using the Swagger UI.
- ✅ Declarative DSL: Define workflows and tasks with simple decorators.
- 🔍 Auto-discovery: Workers automatically discover and register workflow definitions on startup. Zero manual imports required.
- 🕸️ Real DAG Engine: Workflow completion uses networkx graph traversal, correctly handling fan-out and complex topologies.
- ⚡ True Distribution: Each workflow step runs as an independent Celery task, enabling horizontal scaling across workers.
- 🔀 Advanced Flow Control: Support for Fan-out, Branching (conditions), and Join/Barrier (
wait_for). - 🔄 Resilience: Automatic retries with exponential backoff.
- 🏗️ Hexagonal Architecture: Decoupled, testable, and maintainable code.
- 📊 Full Persistence: Every execution state is stored in PostgreSQL.
- 🔭 Distributed Tracing: Full OpenTelemetry instrumentation with Jaeger. Every workflow and step execution is traced end-to-end.
- 🛡️ Saga Pattern: Automatic compensating actions when a step fails. Completed steps are rolled back in reverse order.
- 🔖 Workflow Versioning: Each execution persists the exact workflow version used. Workers resolve the correct definition by version, ensuring in-flight executions are never affected by workflow changes.
- 💻 Native CLI: Interact with Flowcore from the terminal using
flowcore run,flowcore status,flowcore listandflowcore workflows. Rich-formatted tables with color-coded status. - 🏢 Multi-tenancy: Full tenant isolation via X-Tenant-ID header. Each tenant sees only their own executions. Cross-tenant access is blocked at the API level.
Flowcore bridges the gap between simple task queues and heavyweight orchestrators.
| Feature | Flowcore | Celery | Prefect | Airflow | Temporal |
|---|---|---|---|---|---|
| Durable Execution | ✅ | ❌ | ❌ | ✅ | |
| Low Overhead | ✅ | ✅ | ❌ | ❌ | ❌ |
| Native State | ✅ | ❌ | ✅ | ✅ | ✅ |
| Learning Curve | Low | Low | Medium | High | High |
Every technology choice in Flowcore was made deliberately:
- Python 3.11+: Leveraging static typing and performance improvements.
- uv: Ultra-fast dependency management.
- FastAPI: Modern, asynchronous API framework.
- Celery 5.5+: The de-facto standard for distributed tasks in Python.
- SQLAlchemy 2.0: Modern declarative style with async support.
- PostgreSQL: Reliable and relational persistence.
graph TD
API[FastAPI] -->|Start Workflow| Service[WorkflowService]
Service -->|Create Record| DB[(PostgreSQL)]
Service -->|Enqueue| TaskInit[execute_workflow_task]
TaskInit -->|Enqueue Steps| TaskStep[execute_step_task]
TaskStep -->|Loop/Next| TaskStep
TaskStep -->|Update State| DB
The project follows a Hexagonal Architecture (Ports and Adapters):
- Domain: Pure business logic (DSL, Engine).
- Application: Use cases and orchestration.
- Infrastructure: Persistence implementations (SQLAlchemy).
- Adapters: Input/Output ports (FastAPI, Celery).
- WorkflowDefinition: The "blueprint" of the process defined by the user.
- WorkflowExecution: A live instance of a workflow currently running.
- StepExecution: The individual state of each step, with its input/output data.
- Registry: The catalog where definitions reside.
flowcore/
├── src/
│ └── flowcore/
│ ├── domain/ # Pure logic (DSL, Engine)
│ ├── application/ # Use cases
│ ├── infrastructure/ # DB, Repositories
│ └── adapters/ # API, Worker
├── tests/ # Test suite
├── migrations/ # Alembic migrations
├── pyproject.toml # uv configuration
├── Dockerfile # Optimized image
└── Makefile # Automation
- None reported: All core durability and atomicity requirements are now implemented.
- Phase 1 (MVP): Basic orchestration, persistence, and initial DSL. ✅ Completed
- Phase 2: Real distributed step execution. ✅ Completed
- Phase 3: DAG engine with networkx + auto-discovery. ✅ Completed
- Phase 4: Observability with OpenTelemetry + Jaeger. ✅ Completed
- Phase 5: Sagas / Compensating Actions. ✅ Completed
- Phase 6: Native CLI. ✅ Completed
- Phase 7: Workflow versioning. ✅ Completed
- Phase 8: Multi-tenancy. ✅ Completed
Contributions are welcome! Please read CONTRIBUTING.md for more details on how to get started.
Ezequiel Ranieri
Backend & Security Engineer | Distributed Systems & Authentication
📧 ez.ranieri@gmail.com
🐙 GitHub
💼 LinkedIn
MIT License.