Metadata-Version: 2.5
Name: fluid-workflow-engine-sdk
Version: 0.4.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 :: 3 - Alpha
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 (gRPC), 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 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, and heartbeats.
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`, and `scope_path`.

```python
from fluid_workflow_engine_sdk import WorkerClient, RetryableError

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

@worker.step("createThing", description="Creates a thing", default_timeout="30s", max_retries=3)
def create_thing(inputs, log=None, auth=None, **_):
    if log:
        log("INFO", "creating", name=inputs["name"])
    if not_ready():
        raise RetryableError("dependency not ready")   # engine retries per policy
    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: `grpc_port=0` binds an OS-assigned port (the bound
port is what gets registered — required under pre-fork servers), and
`wait_for_engine_s=30` retries engine registration with backoff at startup.
Introspection: `worker.worker_id`, `worker.is_running`, `worker.bound_port`,
`worker.active_executions`, `worker.step_metadata()`.

## 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): construct with `grpc_port=0`, 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()`. Never start in the master under
`--preload` (gRPC servers/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_WORKER_HOST, FLUID_GRPC_PORT,
                                       # FLUID_MAX_WORKERS, FLUID_TEE_LOGGING,
                                       # FLUID_WAIT_FOR_ENGINE_S
worker = WorkerClient.from_settings(settings)
```

Keyword overrides win over the environment; the legacy
`WORKFLOW_ENGINE_ADDRESS` / `SERVICE_HOST` names are honored as deprecated
fallbacks.

## 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.
