Metadata-Version: 2.4
Name: event-bridge-client
Version: 1.2.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'
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"})
```

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

## 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`       | Exponential-backoff retries for event batch POSTs      |
| `nonce_store`       | in-memory | Replay store; supply a shared one for multi-instance   |

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