Metadata-Version: 2.4
Name: event-bridge-client
Version: 1.4.0
Summary: Lightweight client for registering apps, batching lifecycle events, and handling HMAC-signed remote commands.
License: UNLICENSED
Keywords: commands,events,hmac,webhooks
Requires-Python: >=3.9
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.4
Provides-Extra: dev
Requires-Dist: anyio>=4; extra == 'dev'
Requires-Dist: fastapi>=0.110; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == 'fastapi'
Provides-Extra: flask
Requires-Dist: flask>=2; extra == 'flask'
Description-Content-Type: text/markdown

# event-bridge-client (Python)

A lightweight client for connecting a Python application to a control backend:
register the app, batch and push lifecycle events, handle HMAC-signed remote
commands, and take payments. Wire-compatible with the Node `event-bridge-client`
(same version, same signing scheme, same endpoints, same payloads).

## Install

```bash
pip install event-bridge-client          # core (send events, verify commands)
pip install "event-bridge-client[fastapi]"  # + FastAPI inbound adapter
```

Requires Python 3.9+.

## Quickstart (FastAPI)

```python
from fastapi import FastAPI, Request
from event_bridge_client import create_client

client = create_client(
    base_url="https://bridge.example.com",
    api_key="...",
    callback_url="https://api.example.com/bridge/commands",
    callback_secret="...",
    capabilities=["user.ban", "user.unban"],
    env="PROD",
)

@client.on_command("user.ban")
async def _(data, ctx):
    await ban_account(data["externalUserId"], data["reason"])
    return {"ok": True}

app = FastAPI()

@app.post("/bridge/commands")
async def commands(request: Request):
    return await client.middleware.fastapi(request)

@app.on_event("startup")
async def _startup():
    await client.register()

@app.on_event("shutdown")
async def _shutdown():
    await client.aclose()

# Anywhere in your app:
client.events.emit("user.created", {"externalUserId": "usr_123", "email": "a@b.c"})
```

Any event `type` is accepted — emit a custom event the SDK doesn't model and it
is batched and pushed like any other (see `EVENT_TYPES` for the built-ins):

```python
client.events.emit("widget.exported", {"widgetId": "w_1", "rows": 4200})
```

Command handlers may be sync or async. Each receives `(data, ctx)` where `data`
is the command payload (a dict) and `ctx` carries `command_id`, `command_type`,
`issued_at`, and `issued_by`. Return `{"ok": True, "result": ...}` or
`{"ok": False, "error": "..."}`.

### Open dispatch + `"*"` catch-all

Any command `type` is accepted — the envelope is validated, but you can handle a
capability the SDK doesn't model yet with **zero SDK change**. Register a handler
under a specific `type`, or under `"*"` to catch every unregistered type:

```python
@client.on_command("future.capability")          # specific type
async def _(data, ctx): ...

@client.on_command("*")                            # catch-all (anything else)
async def _(data, ctx):
    log.info("got %s", ctx.command_type)           # the actual command type
    return {"ok": True}
```

A specific handler always wins over `"*"`. If neither a specific handler nor a
`"*"` handler is registered, the callback returns **501** (`NO_HANDLER`).

### Known-type validation

For the seven command types the SDK ships typed schemas for — `user.ban`,
`user.unban`, `user.extend_trial`, `subscription.change_plan`,
`subscription.cancel`, `subscription.uncancel`, `subscription.extend`
(see `KNOWN_COMMAND_TYPES` / `COMMAND_DATA_BY_TYPE`) — the `data` payload is
**strictly validated** (and defaults applied, e.g. `atPeriodEnd` → `True`) before
your handler runs. Malformed `data` is rejected with **400** (`invalid_command`)
and your handler is never called. Unknown types pass their `data` through
untouched.

## Payments

Create a payment (returns a hosted checkout URL) and receive the outcome as a
signed webhook on the **same** callback endpoint as commands. Do **not** emit
`payment.*` as events — they're rejected.

```python
# Create → send the payer to hostedCheckoutUrl.
res = await client.payments.create(
    external_id="order_4821",        # your idempotency key
    currency="UZS",
    line_items=[{"name": "Pro plan — 1 month", "unit_amount": 120_000}],
    customer={"email": "alice@example.com", "external_id": "usr_123"},
)
checkout_url = res["hostedCheckoutUrl"]

# Receive the outcome. The same middleware verifies + dispatches here.
@client.on_payment
async def _(note, ctx):
    if note.type == "payment.settled":
        await fulfil(note.data["externalId"])
    elif note.type == "payment.failed":
        await mark_failed(note.data["externalId"])

# (Optional) poll instead of / alongside the webhook:
latest = await client.payments.get_by_external_id("order_4821")  # or .get(id); None if missing
```

Notes:
- `external_id` is your idempotency key — reusing it returns the existing payment.
- Line items take ad-hoc `name` + `unit_amount`, or a catalogue `variant_id`.
- A non-base currency requires `rate_to_uzs`.
- Make `on_payment` **idempotent** — delivery is at-least-once (a raised handler
  returns 500 so management retries). Dedupe on `note.data["paymentId"]` + `note.type`.
- The webhook defaults to your `callback_url`; override per payment with `notify_url`.

## Managed resources

Declare an entity the backend can list / search / view / action — entirely from
the descriptor, with no backend-side code change. Records are never shipped to
the backend; it proxies `list` / `get` / `action` queries back over the same
signed channel.

```python
client.define_resource(
    {
        "key": "widgetUser",
        "label": "Widget User",
        "labelPlural": "Widget Users",
        "titleField": "email",
        "fields": [
            {"key": "id", "label": "ID", "type": "string", "listVisible": False},
            {"key": "email", "label": "Email", "type": "email", "filterable": True},
            {"key": "plan", "label": "Plan", "type": "enum", "enumValues": ["free", "pro"]},
        ],
        "actions": [
            {
                "capability": "widgetUser.ban",
                "label": "Ban",
                "confirm": True,
                "destructive": True,
                "fields": [{"name": "reason", "label": "Reason", "kind": "textarea", "required": True}],
            }
        ],
    },
    # `query` is a ResourceListQuery: query.page, query.page_size, query.q, ...
    list=lambda query: {"records": db.search(query.q, query.page, query.page_size), "total": db.count()},
    get=lambda record_id: db.find(record_id),  # optional
    action=lambda inp: ban(inp["recordId"], inp["params"]["reason"]),  # optional
)
```

Descriptor keys accept either `snake_case` or `camelCase`; they're sent to the
backend as `camelCase`. Field `type` is one of `string`, `number`, `boolean`,
`date`, `datetime`, `enum`, `currency`, `badge`, `email`, `url`, `json`.

## Options

`create_client(...)` keyword arguments:

| Option              | Default   | Notes                                                  |
|---------------------|-----------|--------------------------------------------------------|
| `base_url`          | required  | Control backend base URL                               |
| `api_key`           | required  | API key minted by the backend admin                    |
| `callback_url`      | required  | HTTPS URL the backend POSTs commands to                |
| `callback_secret`   | required  | HMAC shared secret minted alongside the API key        |
| `capabilities`      | `()`      | Strings matching command types, e.g. `user.ban`        |
| `env`               | `"PROD"`  | `PROD` / `STAGING` / `DEV`                              |
| `enabled`           | `True`    | If `False`, all methods are no-ops (staged rollout)    |
| `batch_interval_ms` | `1500`    | Event batcher flush interval                           |
| `batch_max_size`    | `100`     | Force-flush when this many events are queued           |
| `max_buffer_size`   | `10000`   | Hard cap on buffered events; oldest dropped past it    |
| `max_retries`       | `6`       | Retries for one-shot calls + event batch re-queue cap  |
| `request_timeout`   | `15.0`    | Per-request timeout (seconds) for every HTTP call      |
| `heartbeat_interval_ms` | `60000` | Heartbeat cadence                                     |
| `shutdown_timeout`  | `5.0`     | Max seconds to drain buffered events on `aclose()`     |
| `nonce_store`       | in-memory | Replay store; supply a shared one for multi-instance   |

## Reliability: retries, timeouts, shutdown

Every one-shot call (`register`, heartbeat, command `ack`, and `payments.*`) goes
through a shared HTTP core that:

- applies the `request_timeout` per request;
- retries on transport/timeout errors and `408 / 429 / 5xx` with exponential
  backoff + **full jitter** (other `4xx` are never retried), honoring a
  `Retry-After` header when present;
- stamps an `x-request-id` correlation header on every request.

On exhaustion it raises `RequestTimeoutError` (a timeout; `code="TIMEOUT"`,
`status=408`) or `ManagementError` (network error / `HttpError` with
`code="http_<status>"`). The event batcher keeps its own re-queue/backoff and so
issues its POST with retries disabled (it still gets the timeout + correlation
id).

`await client.aclose()` cancels the heartbeat and drains buffered events under
`shutdown_timeout`; if the backend is unreachable it logs a warning and returns
rather than hanging.

## Flask / WSGI

The async `fastapi()` adapter is preferred, but a sync bridge is provided for
WSGI frameworks. Flask is imported lazily, so it stays an optional dependency:

```python
from flask import Flask, request

app = Flask(__name__)

@app.post("/bridge/commands")
def commands():
    return client.middleware.flask(request)   # returns a Flask Response
```

There is also `client.middleware.handle_sync(raw_body, headers, method, path)`
returning `(status, body)` if you build the response yourself.

Constraints (best-effort by design): `handle_sync`/`flask` spin up a short-lived
event loop via `asyncio.run`, so they **must not** be called from inside a running
event loop — use `fastapi()` / `handle` in async apps. Resource queries and
payment webhooks are handled inline and fully supported; command callbacks are
dispatched and acked on that short-lived loop (best-effort).

## Replay protection across instances

The default replay cache is **in-process** — it only protects a single instance.
If you run more than one instance behind a load balancer, supply a shared
`nonce_store` (e.g. Redis) so a captured command can't be replayed against
another instance inside the 300-second signature window:

```python
class RedisNonceStore:
    def __init__(self, redis): self.r = redis
    async def has(self, nonce: str) -> bool:
        return await self.r.exists(f"bridge:nonce:{nonce}") > 0
    async def add(self, nonce: str, ttl_ms: int) -> None:
        await self.r.set(f"bridge:nonce:{nonce}", "1", px=ttl_ms, nx=True)

client = create_client(..., nonce_store=RedisNonceStore(redis))
```

## Send-only usage (no FastAPI)

If the app only emits events and never receives commands, you don't need
FastAPI — `pip install event-bridge-client` and use `register()` /
`events.emit()` / `aclose()`. The middleware is only needed to receive commands.
