Metadata-Version: 2.4
Name: eventpilot
Version: 0.0.1
Summary: EventPilot Python SDK
Author: EventPilot
License: MIT
Project-URL: Homepage, https://github.com/usekairo/eventpilot
Project-URL: Documentation, https://eventpilot.dev
Project-URL: Repository, https://github.com/usekairo/eventpilot
Project-URL: Issues, https://github.com/usekairo/eventpilot/issues
Keywords: eventpilot,workflow,orchestration,event-driven
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: grpcio>=1.62.0
Requires-Dist: aiohttp>=3.9.0
Provides-Extra: typed
Requires-Dist: pydantic>=2.0; extra == "typed"
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.24.0; extra == "otel"
Provides-Extra: test
Requires-Dist: pytest>=8.0.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "test"

# EventPilot Python SDK

Durable, class-based workflow handlers for Python. Production-ready primitives
for retries, rate limiting, cancellation, signals, queries, versioning,
heartbeats, scheduling, and deterministic replay.

Install:

```bash
pip install eventpilot              # core
pip install 'eventpilot[typed]'     # + Pydantic v2 payload validation
pip install 'eventpilot[otel]'      # + OpenTelemetry span emission
```

## Core concepts

- **Handler** — a class decorated with `@handler(name)` and an
  `async def run(self, state)` entrypoint. One class = one workflow type.
- **Step** — an `async def` method decorated with `@step(name)`. Steps are
  the durable unit: each one checkpoints its result before `run()` continues.
- **State** — the `WorkflowState` passed to `run()` and each step. Carries
  the trigger payload, accumulated context, and durability helpers.
- **Closures are never required.** All data flows through `self` and `state`.

## Minimal example

```python
import asyncio
from eventpilot import Worker, handler, step, StepResult, WorkflowState

@handler("order.processing", version="1.0.0", max_retries=5)
class OrderProcessing:
    async def run(self, state: WorkflowState):
        await self.next_step(self.validate)
        await self.next_step(self.charge)

    @step("validate")
    async def validate(self, state: WorkflowState) -> StepResult:
        payload = state.get_payload_map()
        return StepResult(data={"order_id": payload["order_id"]})

    @step("charge", rate_limit_scope="gateway", rate_limit_rate=100)
    async def charge(self, state: WorkflowState) -> StepResult:
        charge_id = await self.side_effect(
            "gateway-charge", lambda: gateway.charge(state.get_string("order_id"))
        )
        return StepResult(data={"charge_id": charge_id})

async def main():
    w = Worker("localhost:50052", http_addr="http://localhost:8080")
    w.register(OrderProcessing)
    await w.start()

asyncio.run(main())
```

## Feature matrix

| Feature | API | Notes |
|---|---|---|
| Retries | `@handler(max_retries=...)` / `@step(retry_policy=...)` | exp/linear/fibonacci/fixed backoff, jitter, non-retryable codes |
| Rate limiting | `@step(rate_limit_scope=, rate_limit_rate=)` | worker-side semaphore |
| Cancellation | `client.cancel_workflow(wf_id, reason=)` | cascade to children via `cancel_policy="CASCADE"` |
| Durable sleep | `await self.sleep(name, duration_ms)` | survives restart |
| Sleep until | `await self.sleep_until(name, deadline)` | absolute UTC |
| Wait for event | `await self.wait_for_event(name, event=, timeout_ms=, correlation_key=)` | with timeout |
| Wait for signal | `await self.wait_for_signal(name, signal=, timeout_ms=)` | sugar over wait_for_event |
| Send signal | `pub.send_signal(wf_id, name, payload)` | wakes `wait_for_signal` / `@signal` |
| Child workflow (sync) | `await self.child_workflow(name, event_name=, payload=, timeout_ms=)` | returns `ChildWorkflowResult` |
| Child workflow (async) | `await self.child_workflow_async(name, event_name=, payload=)` | fire-and-forget |
| Idempotency | `StepResult(side_effect_key=...)` | dedup external side effects across retries |
| Timeouts | `StepTimeouts(schedule_to_start_ms=, start_to_close_ms=, schedule_to_close_ms=, heartbeat_timeout_ms=)` | + `on_timeout=` callback |
| Heartbeat | `await self.heartbeat(progress=...)` | for long-running steps; surfaces as `state.last_progress` on retry |
| Determinism | `await self.now() / self.random() / self.side_effect(key, fn)` | replay-safe time / random / IO |
| Versioning | `await self.get_version("change-id", default=2)` | Temporal-style patched branching |
| Queries | `@query(name)` method | read-only introspection; dispatched when backend support lands |
| Typed signals | `@signal(name, model=Pydantic)` method | declarative handler + payload validation |
| Scheduling | `@scheduled(cron=..., interval_sec=..., ...)` class decorator | auto-registers with orchestrator on worker start |

## Handler authoring: class + methods, no closures

```python
from pydantic import BaseModel
from eventpilot import handler, step, query, signal, scheduled, StepResult, WorkflowState

class OrderPayload(BaseModel):
    order_id: str
    amount: float
    is_vip: bool = False

@scheduled(cron="0 3 * * *", name="nightly-audit")
@handler("order.processing")
class OrderProcessing:

    async def run(self, state: WorkflowState):
        payload = state.parse_payload(OrderPayload)

        await self.next_step(self.validate)

        if await self.get_version("fraud-check", default=1) >= 2:
            await self.next_step(self.fraud_check)

        approved = await self.wait_for_signal("manager-approval", signal="approval", timeout_ms=3_600_000)

        await self.next_step(self.charge)

    @step("validate")
    async def validate(self, state) -> StepResult: ...

    @step("fraud-check")
    async def fraud_check(self, state) -> StepResult: ...

    @step("charge", rate_limit_scope="gateway", rate_limit_rate=100)
    async def charge(self, state) -> StepResult:
        charge_id = await self.side_effect("charge", lambda: gateway.charge(...))
        await self.heartbeat(progress={"stage": "committed"})
        return StepResult(data={"charge_id": charge_id}, side_effect_key="charge")

    @query("get_stage")
    async def get_stage(self, state) -> dict:
        return {"stage": state.context.get("stage", "init")}

    @signal("approval", model=dict)
    async def on_approval(self, state, payload): ...
```

## Observability

```python
from eventpilot import Worker, MetricsRecorder

class PromMetrics:
    def record_step_started(self, handler, step, workflow_id): ...
    def record_step_completed(self, handler, step, workflow_id, duration_ms): ...
    def record_step_failed(self, handler, step, workflow_id, error_code, retryable): ...
    def record_step_deferred(self, handler, step, workflow_id, delay_ms): ...
    def record_workflow_cancelled(self, handler, workflow_id, reason): ...

w = Worker("localhost:50052", metrics_recorder=PromMetrics())
```

If `opentelemetry-api` is installed and a tracer provider is configured,
every step execution is automatically wrapped in an `eventpilot.step.<name>`
span with `eventpilot.{handler,step,workflow_id}` attributes.

Structured logs use `extra={workflow_id, ep_handler, ep_attempt, ep_step}`
— compatible with any JSON log aggregator.

## Clients

```python
# Sync — safe from scripts / non-async code.
from eventpilot import EventPilotClient
client = EventPilotClient(base_url="http://localhost:8080")
client.publish_event("order.created", {"order_id": "ord_123"})

# Async — non-blocking; use from async workers.
from eventpilot import AsyncEventPilotClient
async with AsyncEventPilotClient(base_url="http://localhost:8080") as c:
    await c.publish_event(...)
```

## Testing

```python
from eventpilot import TestEnv, TestStatus, new_state

env = TestEnv()
env.register(OrderProcessing)

# Whole-workflow trigger
result = await env.trigger("order.processing", {"order_id": "ord_1", "amount": 10, "is_vip": False})
assert result.status == TestStatus.COMPLETED

# Query invocation
state = new_state("order.processing").with_context(stage="charging").build()
response = await env.query("order.processing", "get_stage", state)

# Deliver signals / events
env.signal(state.workflow_id, "approval", {"ok": True})
env.deliver_event(state.workflow_id, "payment.succeeded", {"id": "p_1"})

# Inspect heartbeats
assert env.heartbeats
```

## What changed from v0.1

- `@step.compensate` was removed until backend dispatch lands — declare
  rollback logic explicitly or emit a compensation event.
- `EventPilotClient` is now async-aware. Sync calls stay sync; inside an
  async loop use `AsyncEventPilotClient` to avoid blocking.
- Step result context values are strictly JSON-validated — non-JSON types
  now raise `TypeError` at checkpoint time instead of silently corrupting.
- Checkpoints now retry with exponential backoff and raise
  `CheckpointWriteError` on persistent failure, guaranteeing at-least-once
  step delivery.

## License

Apache 2.0 — part of [EventPilot](https://github.com/usekairo/eventpilot).
