Metadata-Version: 2.4
Name: dex-python-sdk
Version: 0.2.6
Requires-Dist: grpcio>=1.83.0
Requires-Dist: grpcio-status>=1.83.0
Requires-Dist: protobuf>=7.35.1
License-File: LICENSE
License-File: LEGACY_NOTICES.md
Summary: Python SDK for the Dex workflow engine
Author: Super Durable
License-Expression: LicenseRef-Super-Durable-1.0
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Homepage, https://github.com/superdurable/dex/tree/main/sdk-python
Project-URL: Repository, https://github.com/superdurable/dex


# Dex SDK for Python

Python SDK for [Dex workflow engine](https://github.com/superdurable/dex)

## New user contracts

The rewrite targets Python 3.11+ and exposes strongly typed workflow contracts
from `dex`. This phase includes definitions, attributes, channels, streams, waits,
decisions, codecs, registry validation, synchronous client calls, and synchronous
worker handlers. Python owns its gRPC Client and Worker transport;
the shared Rust Core is used only for BlobCache.

```python
from datetime import timedelta
from typing import Generator

import dex

counter = dex.Attribute("counter", int)
counters_by_region = dex.AttributeMap("counters-by-region", int)
progress = dex.Stream("progress", str, 10 * 1024 * 1024)

class Run(dex.Step[str]):
    def wait_for(
        self, context: dex.Context, input: str
    ) -> dex.Wait:
        return dex.Wait.until(
            dex.Timer.by_duration(timedelta(seconds=1))
        )

    def execute(
        self, context: dex.Context, input: str
    ) -> Generator[dex.StepOutput, None, dex.StepDecision]:
        yield progress.write(context, "running")
        yield dex.heartbeat({"phase": "running"})
        return dex.graceful_complete(input)

class CounterFlow(dex.Flow[str]):
    run = Run()

    def get_flow_type(self) -> str:
        return "Counter"

    def get_steps(self) -> dex.StepList[str]:
        return dex.StepList.start_step(self.run)

    def get_persistence_schema(self) -> dex.PersistenceSchema:
        return dex.PersistenceSchema.of(counter, counters_by_region, progress)

    @dex.rpc(name="Increment")
    def increment(
        self, context: dex.Context, input: int
    ) -> dex.RPCResult[int]:
        return dex.RPCResult(input + 1)

flow = CounterFlow()
registry = dex.Registry((flow,))
```

Registry derives codecs from declared Python types and handler annotations.
Built-in primitive types and dataclasses need no codec arguments. Register an
explicit codec only for a custom encoding or a type Registry cannot derive.
`PersistenceSchema.of(...)` accepts attributes, channels, and streams together and
partitions them by definition type.

`Worker` and `AsyncWorker` synchronize all registered Indexed Attributes with
Dex Server before opening their listener. Existing indexes return immediately;
failure or the default two-minute deadline aborts startup. An indexed
`AttributeMap` must provide one fixed `index_key`.

Initial attributes retain their value types without a public wrapper class:

```python
options = (
    dex.StartFlowOptions()
    .with_attribute(counter, 1)
    .with_attribute(counters_by_region, "us-west", 1)
)
```

Opt in when declaring an Attribute or AttributeMap, and select the Store in
Flow configuration:

```python
email = dex.Attribute("customer-email", str, sync_to_attribute_store=True)
config = dex.FlowConfig(attribute_store_names=["profiles", "audit"])
```

Stores are asynchronous latest-state projections. Every enabled Attribute write
is sent to every selected Store. Deletion writes SQL `NULL`, and projection
failures do not roll back Flow Attributes. `None` preserves current targets;
an explicit empty list disables future synchronization while retaining protocol
presence.

```
pip install dex-python-sdk==0.1.0
```

See [samples](../examples/python) for use case examples.

## Requirements

- Python 3.11+
- [Dex server](https://github.com/superdurable/dex#how-to-use)

## Concepts

Applications implement two generic interfaces from [`dex`](dex/):

- `Flow[START_INPUT]` returns `StepList.start_step(...)`, followed by optional
  `.other_steps(...)`, from one `get_steps()` method. The `StepList` generic
  binds the Flow input to the starting Step input. Use `StepList.empty()` when
  a Flow has no Steps.
- `Step[INPUT]` implements `execute` and optionally `wait_for`. The default
  Worker accepts ordinary synchronous handlers and generator handlers. A generator
  yields `StepOutput` progress frames and returns its final `Wait` or `StepDecision`.
  With `AsyncWorker` and `Registry(..., allow_async_handlers=True)`, Step
  coroutines use `AsyncContext`; async RPCs keep `Context`.

`StepOptions.wait_for_method_timeout` and `execute_method_timeout` bound the
two handler calls. Timer and channel conditions determine how long a Step waits.

`wait_for_retry` and `execute_retry` limit one logical handler execution. With
`StepDurability.ASYNC`, local and fallback regular activities share maximum
attempts, total duration, and 1-based attempt numbers. Fallback starts
immediately; later regular retries continue the backoff sequence at the
cumulative attempt.

The default Step durability is synchronous. A Flow configuration can select
asynchronous durability, and a Step method override has highest precedence. The
default retry total duration is four hours. Regular attempts default to a two-hour
method timeout and one-minute heartbeat timeout; an explicit heartbeat timeout must
meet the server minimum, which defaults to ten seconds. Asynchronous durability
first allows at most three local attempts in seven seconds. The local phase ignores
method timeouts and heartbeat frames before falling back to a regular activity.

### Step progress and heartbeat recovery

A synchronous handler yields every heartbeat and Stream write. The generator return
value is the only final result:

```python
def execute(
    self, context: dex.Context, input: str
) -> Generator[dex.StepOutput, None, dex.StepDecision]:
    yield dex.heartbeat({"offset": 10})
    yield progress.write(context, "processed 10 items")
    return dex.graceful_complete(input)
```

An asynchronous handler keeps its normal coroutine return. Stream writes enqueue
without waiting for Stream Store acknowledgement; heartbeat waits only for the
Worker output queue:

```python
async def execute(
    self, context: dex.AsyncContext, input: str
) -> dex.StepDecision:
    progress.write(context, "started")
    await context.heartbeat({"offset": 10})
    return dex.graceful_complete(input)
```

Call `heartbeat()` or `await context.heartbeat()` without a value to clear previous
details. Passing Python `None` persists a present null Value. On a later regular
attempt, use `context.has_last_heartbeat_value()` before decoding with
`context.get_last_heartbeat_value(ExpectedType)`. A Stream frame is also an implicit
heartbeat, but it preserves the latest explicit heartbeat state.

A Step may write the same Stream any number of times. Dex assigns
`#<stepExecutionID>` as each Step message's `StreamMessage.source`. Client writes
provide their own non-empty source; duplicate sources and `#` are allowed and every
write appends:

```python
client.write_stream(flow_id, progress, "frontend#preview", "rendering")
message = client.read_stream(flow_id, progress)
print(message.source)
```

### Canceling Step executions

A successful Step can cancel queued or active executions while continuing with
its normal decision:

```python
return (
    dex.go_to(RecordQuote, quote)
    .with_canceling_sibling_steps(QuoteCarrierA, QuoteCarrierB)
    .with_canceling_steps(GlobalQuoteTimeout)
)
```

`with_canceling_steps` selects every current execution of each registered Step
type. `with_canceling_sibling_steps` selects only executions with the same
`Context.from_step_execution_id` as the current execution. Decisions are
immutable; repeated calls form a union, and Flow-wide selection wins for the
same Step type. Unregistered selectors produce an invalid Step result.

Dex resolves one snapshot after the current execution succeeds. Completed,
already-canceled, and absent targets are no-ops. Next Steps created by the same
decision are outside the snapshot. Dex immediately applies the next or close
action; late decisions, writes, retries, and recovery Steps are discarded.

`RPCResult.with_canceling_steps` provides the Flow-wide selector for RPCs.
RPCs do not support sibling selection because they have no Step execution
lineage.

### Soft Flow timeout

Override `Flow.handle_timeout` to make a positive timeout use handler policy
by default. Both synchronous and async Workers support the hook:

```python
class Orders(dex.Flow[str]):
    async def handle_timeout(self, context: dex.Context) -> dex.StepDecision:
        await notify_expiration(context)
        return dex.force_complete("expired")

options = dex.StartFlowOptions(
    timeout=timedelta(minutes=30),
    timeout_policy=dex.FlowTimeoutPolicy.HANDLER,
)
```

Register async hooks with `allow_async_handlers=True` and run them with
`AsyncWorker`. `FAIL` produces `FlowErrorType.FLOW_TIMEOUT` and permits Flow
retry; `CANCEL` cancels without retry. Continue-as-new preserves the deadline,
while retry runs receive a fresh budget. A zero or absent timeout
disables the feature.

`Registry` validates every Flow, Step, RPC signature, durable name, lock, and
codec before Client or Worker startup. `Client` methods use these typed objects
instead of raw Flow, Step, or RPC strings.

### Waiting and map inspection

`Wait.all_of` and `Wait.any_of` may use unnamed Conditions. Every Condition in
`Wait.any_combination_of` must have a non-empty user ID; the same Condition
instance may appear in multiple combinations.

Both `Client` and `AsyncClient` provide singleton and AttributeMap-instance
overloads of `wait_for_attribute_equal`. They target the current run and accept
only string, bool, int, or float wire values. JSON objects, bytes, and null fail
before transport. `AttributeMap.get_map_size/get_all_instance_keys` include
buffered sets and deletes. The matching `ChannelMap` methods are RPC-only,
include buffered publishes, and omit empty instances. Keys are decoded and
sorted. Use `force_complete_if_channels_empty(...)` for conditional completion.

`Client.wait_for_flow` and `AsyncClient.wait_for_flow` return a
`FlowResult` after hydrating every output-bearing completion. Use
`single_output` only when the Flow contract produces exactly one output:

```python
output = client.wait_for_flow(flow_id).single_output(OrderResult)

result = client.wait_for_flow(flow_id)
for completion in result.completions:
    if completion.step_execution_id == expected_execution_id:
        output = completion.decode(OrderResult)
```

`completions` is an immutable tuple in server collection order. Parallel branch
order is not deterministic, so select by `step_type` or `step_execution_id`.
No-output Flows return an empty tuple; `single_output` raises `ValueError` for
zero or multiple completions. Every terminal status returns a `FlowResult`; inspect
`status`, `error_type`, and `error_message` for unsuccessful completion.

SubFlows are normal, independently addressable Flows used as durable Conditions:

```python
def wait_for(self, context: Context, input: ChargeInput) -> Wait:
    return Wait.until(SubFlow.run(self.charge_flow, input))

def execute(self, context: Context, input: ChargeInput) -> StepDecision:
    del input
    receipt = SubFlow.get_condition_results(context).single_output(Receipt)
    return graceful_complete(receipt)
```

`SubFlow.get_flow_id(context, index=0)` remains available for a running `any_of`
loser. `SubFlowOptions` configures timing, timeout policy, retry, initial target
Attributes, Flow config, Condition ID, and reuse. Parent completion does not cancel an unfinished
SubFlow.

### Errors

Client calls raise concrete `DexServiceError` subclasses. Existing-Flow reads
(`get_attribute`, `describe_flow`, `wait_for_flow`, and `time_travel`) raise
`FlowNotFoundError` when the Flow does not exist. Mutations, RPCs, timer/Step
waits, config updates, and continue-as-new triggers raise
`FlowNotActiveError` when no running Flow can accept the operation.

```python
try:
    client.publish(flow_id, orders.approved, order_id)
except dex.FlowNotActiveError:
    # The Flow is missing or already closed.
    pass
```

Duplicate starts, worker failures, RPC lock contention, and long-poll timeouts
raise `FlowAlreadyStartedError`, `WorkerInvocationError`,
`RpcLockConflictError`, and `LongPollTimeoutError`. All service errors retain
`code`, `sub_status`, `detail`, `operation`, `flow_id`, and the original gRPC
exception through Python exception chaining. Worker failures also expose
`worker_code`, `worker_error_type`, and `worker_error_detail`. Registration,
serialization, and invalid handler returns use `FlowDefinitionError`,
`ValueMappingError`, and `InvalidStepResultError`.

### Sync vs asyncio

- **Sync (default):** `Client` and `Worker` use blocking gRPC and a thread-pool
  Worker. A progress generator cooperatively hands each yielded frame to gRPC;
  `Stream.write` must therefore be yielded. Blocking Client calls inside
  `Step.execute` are safe while other pool threads remain available.
- **Asyncio:** `AsyncClient` and `AsyncWorker` use `grpc.aio`. Use
  `Registry(..., allow_async_handlers=True)` when Steps/RPCs are coroutines.
  Step coroutines annotate `AsyncContext`, call `Stream.write` without `await`,
  and await `context.heartbeat`. Inside async `execute`, inject `AsyncClient` —
  do not call sync `Client` on the Worker event loop. Async generators are rejected.

Integration scenarios live under
[`tests/integ`](tests/integ/README.md). They exercise the same workflows,
client operations, and assertions as the Java suite against an isolated
`dexcli dev` environment.

## Implementation status

The strongly typed contracts, registry, synchronous Client/Worker, optional
`AsyncClient`/`AsyncWorker` (`grpc.aio`), and Rust-backed BlobCache are
implemented. Python owns its gRPC transport; the native bridge is limited to
the shared BlobCache. Design notes:
[`python-sdk-async-apis.md`](../docs/design/plan/python-sdk-async-apis.md) and
[`python-sdk-step-streaming.md`](../docs/design/plan/python-sdk-step-streaming.md).

## Running Dex locally

Install and start the complete local environment with `dexcli`:

```bash
brew install superdurable/tap/dexcli
dexcli dev
```

Dex Server listens on `127.0.0.1:8801`. See the
[CLI README](../cli/README.md) for endpoints and persistence options.

## How To Contribute

This project uses [uv](https://docs.astral.sh/uv/) for Python versions,
dependencies, virtual environments, locking, building, and publishing.

To install requirements:

```bash
uv sync --locked
```

Run the complete Python SDK integration suite with an isolated Dex development
environment:

```bash
./run-integration-tests.sh
```

### Measure integration coverage

Run the same integration suite with Python source coverage:

```bash
./run-integration-tests.sh --coverage
```

Only the integration scenarios contribute execution data, and only production
Python modules under `dex` are measured. Generated protobuf modules under
`dex/dexpb` are excluded. The
terminal report lists uncovered line ranges. The browser report starts at
`coverage/html/index.html`; `coverage/coverage.xml` and `coverage/lcov.info`
are also generated.

CI uploads LCOV to Codecov with GitHub OIDC under the
`sdk-python-integration` flag and retains the full report as the
`sdk-python-integration-coverage` Actions artifact.

#### Update IDL

Edit [`protos/dex.proto`](../protos/dex.proto). Rename catalog: [`docs/design/idl-renames.md`](../docs/design/idl-renames.md).

#### Generate stubs from IDL

```bash
make -C ../protos proto-python
```

Checked-in Python stubs land in `dex/dexpb/`.
#### Linting

Validate that every `dex.__all__` class, function, constant, public method,
argument, return value, dataclass field, enum value, and public instance
attribute has a Google-style docstring:

```bash
uv run --frozen python scripts/check_public_docs.py
```

The checker resolves definitions from the public package export table, so
private helpers and generated protobuf modules are excluded. Use `help(dex.Client)`
or IDE hover information to read the same documentation. To run all other
linting for this project:

```bash
uv run --frozen pre-commit run --show-diff-on-failure --color=always --all-files
```

## Code of Conduct
This project is governed by the [Contributor Covenant v 1.4.1](CODE_OF_CONDUCT.md). (Review the Code of Conduct and remove this sentence before publishing your project.)

## Publishing to PyPI

1. Optionally run **Publish Python SDK to PyPI** via workflow_dispatch with a version and
   `publish=false` to validate all distributions without uploading.
2. Create a GitHub Release with tag `sdk-python/vX.Y.Z` (for example `sdk-python/v0.1.0`).
   CI stamps that version into `pyproject.toml` for the build (same idea as the TypeScript
   SDK release), then builds and smoke-tests Linux x86_64/ARM64, macOS x86_64/ARM64, and
   Windows x86_64 wheels, verifies the source distribution, and publishes with `PYPI_TOKEN`.
3. After publishing, bump the committed `pyproject.toml` / docs install line when you want
   the repo tip to reflect the released version.

A manual run publishes only from `main`, and only when `publish` is explicitly selected.
The dispatch `version` input is stamped the same way as a release tag.

See [CONTRIBUTING.md](../CONTRIBUTING.md#releases-monorepo-tags) for monorepo tag conventions.

## License

[Super Durable Source License 1.0](LICENSE), with legacy portions under their
original terms as described in [LEGACY_NOTICES.md](LEGACY_NOTICES.md).

