Metadata-Version: 2.5
Name: fluid-workflow-engine-sdk
Version: 1.0.0
Summary: Python SDK for the Coredgeio Fluid workflow engine — worker registration, step execution, and workflow management
Project-URL: Homepage, https://github.com/coredgeio/fluid-workflow-engine
Project-URL: Repository, https://github.com/coredgeio/fluid-workflow-engine
Project-URL: Issues, https://github.com/coredgeio/fluid-workflow-engine/issues
Author-email: "Coredge.io" <ashok@coredge.io>
License: Apache-2.0
Keywords: fastapi,grpc,orchestration,saga,workflow
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Framework :: Flask
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: grpcio>=1.60
Requires-Dist: httpx>=0.27
Requires-Dist: protobuf>=6.31.1
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Provides-Extra: dev
Requires-Dist: fastapi>=0.110; extra == 'dev'
Requires-Dist: flask>=2.3; extra == 'dev'
Requires-Dist: grpcio-tools~=1.74.0; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.30; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: uvicorn>=0.29; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == 'fastapi'
Provides-Extra: flask
Requires-Dist: flask>=2.3; extra == 'flask'
Description-Content-Type: text/markdown

# fluid-workflow-engine-sdk

Python SDK for the [Coredge Fluid workflow engine](https://github.com/coredgeio/fluid-workflow-engine):
remote step workers (outbound gRPC stream — no inbound port), workflow definition
building (YAML, `workflow/v1` + `workflow/v2` flow control), workflow/trigger
registration, and execution management.

```bash
pip install fluid-workflow-engine-sdk            # core
pip install "fluid-workflow-engine-sdk[fastapi]" # + FastAPI integration
pip install "fluid-workflow-engine-sdk[flask]"   # + Flask integration
```

```python
import fluid_workflow_engine_sdk
```

## Migrating to 1.0 (worker-initiated step stream)

1.0 is a hard cutover to the engine's `StepStream` dispatch model: the engine
no longer dials workers. A worker registers, then opens one long-lived
bidirectional stream to the engine's registration address over which the
engine pushes step tasks and the worker sends log/result frames back. The
old `StepWorkerService` gRPC server the SDK used to host is gone, and with it
every inbound-connectivity knob:

- `WorkerClient(..., worker_host=..., grpc_port=...)` → removed. The
  constructor is now `WorkerClient(service_name, engine_address, *,
  max_workers=10, tee_logging=True, wait_for_engine_s=0.0, max_inflight=0)`;
  everything after `engine_address` is keyword-only so a stale positional
  `worker_host` fails at construction.
- `WorkerSettings.worker_host` / `grpc_port`, `FLUID_WORKER_HOST` /
  `FLUID_GRPC_PORT` and the `SERVICE_HOST` alias → removed; `max_inflight` /
  `FLUID_MAX_INFLIGHT` added. Unknown settings fields are now rejected.
- `worker.bound_port` → removed; `worker.attached` added (is the step stream
  open right now). `/fluid/healthz` reports `attached` and lists the in-flight
  task ids under `active_executions`.
- Steps gain an optional `cancel` kwarg (a `threading.Event` the engine sets
  on cancel / timeout / stream loss) and `delivery_count`.
- No Kubernetes `Service`, no `WORKER_HOST`, no pre-fork port juggling: each
  process (including each Gunicorn / uvicorn child) simply opens its own
  stream after the fork.

Requires an engine that serves `WorkflowRegistrationService.StepStream`.

## Migrating from `workflow-engine-sdk` (≤ 0.2.1)

The distribution was renamed `workflow-engine-sdk` → `fluid-workflow-engine-sdk`
and the import package `workflow_engine_sdk` → `fluid_workflow_engine_sdk` in 0.3.0.

- `import workflow_engine_sdk` still works via a deprecated shim (emits
  `DeprecationWarning`); update imports at your convenience.
- **Replace** the old requirement line — never install both distributions in one
  environment. Both own the `workflow_engine_sdk/` path; pip will silently
  clobber files and uninstalling either breaks the other.

```diff
- workflow_engine_sdk==0.2.1
+ fluid-workflow-engine-sdk==0.3.0
```

## Worker quick start

A worker hosts step functions, registers them with the engine, opens the step
stream the engine pushes tasks down, and heartbeats. It never listens for the
engine: the only network requirement is that it can reach `engine_address`.
Steps are plain callables; declare only the keyword args you need — `inputs`,
`workflow_id`, `workflow_name`, `step_name`, `is_compensation`, `retry_count`,
and the optional signature-gated extras `log`, `auth`, `scope_path`, `cancel`
and `delivery_count`.

```python
from fluid_workflow_engine_sdk import WorkerClient, RetryableError

worker = WorkerClient(
    service_name="my-svc",
    engine_address="workflow-engine:50052",
)

@worker.step("createThing", description="Creates a thing", default_timeout="30s", max_retries=3)
def create_thing(inputs, log=None, auth=None, cancel=None, **_):
    if log:
        log("INFO", "creating", name=inputs["name"])
    if not_ready():
        raise RetryableError("dependency not ready")   # engine retries per policy
    if cancel is not None and cancel.is_set():          # engine cancelled / timed out
        raise RuntimeError("cancelled")
    return {"id": "thing-123"}                          # step outputs

@worker.step("deleteThing", supports_compensation=True)
def delete_thing(inputs, is_compensation, **_):
    ...
    return {}

worker.start()      # or: app = FastAPI(lifespan=worker.lifespan)
```

Anything logged with the stdlib `logging` module inside a step body is also
streamed to the engine's event log (disable with `WorkerClient(...,
tee_logging=False)`). Caveat: threads spawned *inside* a step body are not
captured — pass `log` explicitly there.

Steps may be `async def` — each invocation runs on a fresh event loop on the
worker's thread pool (`asyncio.run`), so `log`/`auth` and the logging tee work
unchanged. Don't cache loop-bound resources (e.g. a module-level
`httpx.AsyncClient`) across invocations; create them inside the step.

Useful constructor extras: `wait_for_engine_s=30` retries engine registration
with backoff at startup, `max_workers` sizes the thread pool tasks run on, and
`max_inflight` caps how many tasks the engine may push at once (0 = engine
default). Introspection: `worker.worker_id`, `worker.is_running` (registered),
`worker.attached` (step stream open), `worker.active_executions` (task ids in
flight), `worker.step_metadata()`.

### How dispatch works

```
worker ──RegisterWorker──▶ engine :50052      (steps + metadata, then Heartbeat)
worker ──StepStream (bidi)──▶ engine :50052   (hello → welcome, then:)
        ◀── StepTask / StepTaskCancel
        ──▶ StepLogFrame* , StepTaskResult
```

The stream is worker-initiated and long-lived; if it drops (engine restart,
network blip) the engine immediately requeues every task this worker held to
another worker, the SDK cancels the in-flight steps (their `cancel` event is
set and their results are discarded) and reconnects with exponential backoff
(1s → 30s), re-sending its hello. Step functions should therefore be
idempotent or honour `cancel`; a task's `delivery_count` is > 1 when it was
redelivered after a previous worker lost its stream.

Timeouts: when the engine sets a timeout on a task, the SDK reports a
retryable `timed out` failure at the deadline and sets `cancel`; a step that
ignores `cancel` keeps running in the background but its result is discarded.

## FastAPI integration

```python
from fastapi import FastAPI
from fluid_workflow_engine_sdk.contrib.fastapi import worker_lifespan, worker_router

app = FastAPI(lifespan=worker_lifespan(worker))       # starts/stops the worker
app.include_router(worker_router(worker))             # GET /fluid/healthz, /fluid/steps
```

`worker_lifespan` runs the blocking start/stop off the event loop and composes
with an existing lifespan: `worker_lifespan(worker, inner=app_lifespan)`
(worker start → inner enter → serve → inner exit → worker stop).

## Flask integration

```python
from flask import Flask
from fluid_workflow_engine_sdk.contrib.flask import FlaskWorker

app = Flask(__name__)
FlaskWorker(worker, app)   # starts the worker now, stops it atexit,
                           # mounts /fluid/healthz + /fluid/steps
```

Under a pre-fork server (Gunicorn): call
`FlaskWorker(worker).init_app(app, start=False)` at import, and start each
fork's worker in a `post_fork` hook —
`app.extensions["fluid_worker"].start()` — so every child opens its own step
stream. Never start in the master under `--preload` (gRPC channels don't
survive `fork()`). See [`examples/gunicorn.conf.py`](examples/gunicorn.conf.py).
The Flask dev-server reloader imports the app twice — run with
`use_reloader=False`.

## Step routers

`StepRouter` decouples step declaration from the worker instance, like
FastAPI's `APIRouter` — feature modules own their steps, `main` assembles:

```python
from fluid_workflow_engine_sdk import StepRouter

billing = StepRouter(prefix="billing")   # advertised as "billing.<name>"

@billing.step("charge")
async def charge(inputs, **_):
    return {"chargeId": "..."}

worker.include_router(billing)           # ValueError on duplicate step names
```

Routers nest (`router.include_router(other)`) and can carry workflows
(`router.workflow(defn)`), which register when the including worker starts.

## Workflow auto-registration

```python
worker.workflow(wf)                # WorkflowDefinition, YAML str, or bytes
worker.start()                     # registers steps, then pushes workflows
```

Definitions are pushed right after worker registration
(`source_service=service_name`); a rejected definition raises `EngineError`
and aborts startup. Steps whose `function` is served by this worker get
`executionMode: grpc` defaulted in automatically (the engine dispatches
remotely only on an explicit non-local mode); builtins and other services'
functions are left untouched.

## Configuration via environment

```python
from fluid_workflow_engine_sdk import WorkerClient, WorkerSettings

settings = WorkerSettings.from_env()   # FLUID_SERVICE_NAME, FLUID_ENGINE_ADDRESS,
                                       # FLUID_MAX_WORKERS, FLUID_TEE_LOGGING,
                                       # FLUID_WAIT_FOR_ENGINE_S, FLUID_MAX_INFLIGHT
worker = WorkerClient.from_settings(settings)
```

Keyword overrides win over the environment; the legacy
`WORKFLOW_ENGINE_ADDRESS` name is honored as a deprecated fallback.

## Building workflow definitions

```python
from fluid_workflow_engine_sdk import WorkflowDefinition, Step, RetryPolicy

wf = (
    WorkflowDefinition("provision-fleet")
    .api_version("workflow/v2")            # required for flow-control operators
    .input("regions", type="list", required=True)
    .input("env", type="string", default="staging")
    .step(
        Step("deployAll")
        .foreach("inputs.regions", as_="region", parallel=True, max_concurrency=4)
        .body(
            Step("provision", function="createVpc")
            .input("region", "{{ loop.region }}")
            .retry(RetryPolicy(max_retries=3, initial_backoff="1s"))
            .compensation("deleteVpc", inputs={"region": "{{ loop.region }}"}),
        )
    )
    .step(
        Step("notify")
        .if_("inputs.env == 'prod'")
        .body(Step("page", function="pageOncall"))
        .else_(Step("slack", function="notifySlack"))
        .depends_on("deployAll")
    )
    .output("done", "{{ steps.deployAll.outputs.completed }}")
)
print(wf.to_yaml())
```

`while_`/`until` (polling loops, `max_iterations` mandatory) and
`switch`/`case`/`default` are also available. Step inputs support the object
form for optional values: `Step(...).input("size", "{{ inputs.size }}",
required=False, default="m5.large")`.

## Engine client

```python
from fluid_workflow_engine_sdk import WorkflowEngineClient, TriggerSpec

with WorkflowEngineClient("http://engine:50051", engine_grpc="engine:50052") as client:
    client.register_workflow(wf, replace=True, source_service="my-svc")
    res = client.start_workflow(
        "provision-fleet",
        {"regions": ["us-east-1"]},
        started_by="user:ashok",
        tenant="acme", domain="default", project="demo",
    )
    detail = client.get_execution(res.workflow_id)

    client.register_trigger(
        TriggerSpec(
            name="on-vm-delete",
            resource_type="compute",
            event_type="deleted",
            workflow_name="cleanup-vm",
            input_mappings={"vmName": "event.resource_name"},
        ),
        source_service="my-svc",
        replace=True,
    )
```

Also available: `unregister_workflow`, `list_workflow_definitions` (gRPC
registry view), `resume_from_step(workflow_id, step_name)`,
`unregister_trigger`, `list_triggers`, plus the HTTP admin surface
(`list_workflows`, `get_workflow`, `list_executions`, `get_execution`,
`cancel_execution`).

## Examples

See [`examples/`](examples/) for runnable scripts: basic worker, FastAPI
(plain + router/settings/composed-lifespan), Flask (+ Gunicorn config),
workflow auto-registration, definition building with flow control,
register-and-start, and trigger registration. Framework guides:
[`docs/PYTHON_FASTAPI_WORKER.md`](docs/PYTHON_FASTAPI_WORKER.md) and
[`docs/PYTHON_FLASK_WORKER.md`](docs/PYTHON_FLASK_WORKER.md).

## Development

```bash
pip install -e ".[dev]"
make gen-stubs         # regenerate gRPC stubs from ../api/workflow/workflow_service.proto
make test              # unit + offline integration tests (FakeEngine, no Docker)
make test-integration  # just the offline FakeEngine integration tier
make e2e-up            # MongoDB via docker compose (needs `make build` at repo root)
make test-e2e          # real engine + Mongo end-to-end tier
make e2e-down
```

`grpcio-tools` is pinned so regenerated stubs keep the same protobuf gencode
version as the committed ones (protobuf 6.31.x); if you bump it, bump the
`protobuf` runtime floor in `pyproject.toml` to match the new gencode
requirement.
