Metadata-Version: 2.5
Name: industream-eventhub-client
Version: 1.0.0rc6
Summary: EventHub client for Python: publishes and subscribes to Industream events stored in NATS JetStream, and acknowledges and ends them through the commands of the EventHub backend.
Project-URL: Repository, https://github.com/industream/industream-event-hub
Author: Sullivan Vanet
License-Expression: MIT
Keywords: cloudevents,eventhub,events,industream,jetstream,nats
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: nats-py<3,>=2.16
Description-Content-Type: text/markdown

# industream-eventhub-client

Python client for EventHub: publishes and subscribes to events stored in NATS JetStream, sends the commands that
modify them to the EventHub backend, and reads their history from it.

It is a translation of the .NET client
([Industream.EventHub.Models](https://www.nuget.org/packages/Industream.EventHub.Models) and
[Industream.EventHub.Client](https://www.nuget.org/packages/Industream.EventHub.Client)), as close as possible: same
models, same names, same behaviour. Only the Python conventions change (snake_case members, UPPER_CASE constants and
enum members, no `Async` suffix, `Error` instead of `Exception`). Its models follow the shared schemas and its tests
replay the shared contract cases (`Shared/README.md` in the EventHub repository).

## Installation

```bash
pip install industream-eventhub-client
```

Python 3.12 or later. The client is asynchronous (asyncio) and works on the NATS connection the application opens with
[nats-py](https://github.com/nats-io/nats.py).

## Building an event

```python
from industream.eventhub.client import EventCategory, EventFactory, EventModel, EventSeverity

thermal_alarm = EventModel(
    application="raceway",  # type industream.raceway.thermalalarm
    name="thermalalarm",
    source="/industream/flowmaker/raceway-flow",  # optional, default: /industream/raceway
    severity=EventSeverity.ALARM,
    category=EventCategory.PROCESS,
    subject="tuyere/12",
    attributes={"tuyere": 12},
    data={"code": "THERMAL_OUT_OF_RANGE", "flameTemperature": 2412},
)

cloud_event = EventFactory.create(thermal_alarm)
```

`EventFactory.create` fills in:

- `id` (UUID v7), `specversion` 1.0, `datacontenttype` `application/json`;
- `type` (`industream.<application>.<name>`) and `source` (`source`, or `/industream/<application>` by default);
- `time` in UTC, to the millisecond (now when the model has no `time`);
- the Severity extension (`severitytext` and `severitynumber`), `category`, and `project` (`application` unless the
  model sets `project`);
- `rootevent` when the model sets `root_event_id` (secondary events);
- `keephistory: false` when the model sets `keep_in_history=False`: the event goes through EventHub as usual, but the
  backend leaves it out of the database history (an ordinary event does not carry the extension);
- `lasting: true` when the model sets `lasting=True`: the event lasts until it is ended, like an alarm (an
  instantaneous event does not carry the extension);
- the publisher attributes, keeping their type: `str`, 32-bit `int`, `bool` or `datetime` (written as a timestamp).

`data` is any value `json` can write (`dict`, `list`, `str`, numbers…); a dataclass is written with its field names in
camelCase (`flame_temperature` becomes `flameTemperature`), like the typed data of the .NET client, and a `datetime`
(ISO 8601) or a `UUID` as a string. Another type (an `Enum`, a `Decimal`…) raises a `TypeError` when the event is
written, at `publish`.
`EventJson.encode(cloud_event)` writes the CloudEvent as structured JSON (UTF-8 bytes), as it is published.

## Invalid models

An invalid model is a programming error: `EventFactory.create` raises a `ValueError` whose message says what is wrong:
empty application, event or project name, unknown severity or category, empty subject, source that is not a URI
reference, invalid or reserved attribute name, unsupported attribute value, root event id that is not a `UUID`,
`time` (or `datetime` attribute) without time zone or out of the UTC range, string attribute (subject, names, project,
attribute value…) with a control character, a noncharacter or a lone surrogate.

## Decoding a received event

```python
received_event = EventDecoder.decode(message_body)  # bytes; raises EventDecodeError if invalid
```

`ReceivedEvent` gives the CloudEvents fields, the severity, the category, the project, `root_event_id` (a `UUID`),
`keep_in_history` (`True` when `keephistory` is absent), `lasting` (`False` when `lasting` is absent), the publisher
attributes (read-only) and `data` (as parsed
JSON), plus the received `cloud_event` and, when a client read it from the stream, its `stream_sequence` and
`nats_subject`; a missing value is `None`. `time` is a `datetime` in UTC. The severity is read by OpenTelemetry range (`severitynumber` 13 to 16
is `WARN`, TRACE 1 to 4 is read as `DEBUG`), or from `severitytext` when there is no number: an event published
without the EventHub client may use any number of a level. Publisher attributes keep numbers and booleans typed;
timestamps and URIs, plain strings in JSON, come back as strings.

A message is rejected when it is not a valid CloudEvent in structured JSON (a body starting with a UTF-8 byte order
mark, a blank `id`, `type` or `source`, an
empty `subject`, a `time` that is not an RFC 3339 timestamp of a real day or has an offset beyond 14 hours, an invalid
`datacontenttype` or
`dataschema`, both `data` and `data_base64`, an extension name that is not lowercase letters and digits, a control
character or a noncharacter in a string attribute included),
has no time, severity, category or project, has a `severitynumber` that is not an integer, a `keephistory` or a `lasting`
that is not a boolean, a `rootevent` that is not an event id in its standard form, or an extension value outside the CloudEvents
type system (an object, a list, a number that is not a 32-bit integer). A `null` attribute is an absent one.

### Secondary events

When the backend acknowledges or ends an event, it publishes a **secondary event**: its `root_event_id` is the id of
the main event, its type `industream.eventhub.event.<change>`. A subscriber receives it like any other event, with
`modification` set: what changed and the full state of the main event, read from its data. `modification` is `None`
for every other event.

```python
if received_event.modification is not None:
    if received_event.modification.change == EventChange.ACKNOWLEDGED:
        ...
    end = received_event.modification.state.end  # the full, cumulative state
```

`change` is a string: `EventChange.ACKNOWLEDGED`, `EventChange.ENDED`, or a change added later, kept as received. A
secondary event whose data is not `{ change, event, state }` with the change of its type and a valid state is rejected
like any other invalid event. The main event stays in `data`, as it was published.

## Reading the state of a modified event

`EventStateDecoder.decode(json)` reads the state the backend gives a modified event (Docs/Event-Model.md §8.3), from
parsed JSON: the `state` of a value of the `eventhub-latest` bucket, of a secondary event or of a command reply.

```python
state = EventStateDecoder.decode(value["state"])
# state.acknowledgement: Acknowledgement(actor=Actor(type, id, proof), time, comment) or None
# state.end: EventEnd(actor, time) or None
```

The values are read in kebab-case (`"hub-token"` gives `ActorProof.HUB_TOKEN`), the times as RFC 3339 times
(`datetime` with their offset), and a field it does not know is ignored. An invalid state raises an `EventDecodeError`
that names the wrong field (`'acknowledgement.actor.type' is unknown: 'robot'`).

A value of the `eventhub-latest` bucket (`EventContract.LATEST_VERSION_BUCKET`), the latest version of each modified
event, is read with `LatestEventVersionDecoder.decode(json)`: `event`, the main event decoded as `EventDecoder` does
with its `nats_subject`, `nats_subject` and `state`. The pending secondary event the backend may keep there is
ignored.

```python
bucket = await connection.jetstream().key_value(EventContract.LATEST_VERSION_BUCKET)
entry = await bucket.get(event_id)
latest_version = LatestEventVersionDecoder.decode(json.loads(entry.value))
```

## NATS subject

The publisher chooses the NATS subject of its events; it only has to start with `eventhub.`, the root the EventHub
stream captures (why: `Docs/Event-Model.md` §7.2).

```python
from industream.eventhub.client import EventSubject

EventSubject.validate("eventhub.raceway.process.alarm.thermalalarm")  # raises ValueError if invalid
```

A subject is rejected when it does not start with `eventhub.`, has an empty token, contains whitespace or contains a
wildcard (`*`, `>`: subscriptions only). `EventSubject.ALL` (`eventhub.>`) covers every EventHub subject.

A subscription filter accepts wildcards as whole tokens, `>` only last:

```python
EventSubject.validate_filter("eventhub.*.*.alarm.>")
```

## Publishing an event

```python
import nats

from industream.eventhub.client import EventFactory, EventPublisher

connection = await nats.connect("nats://localhost:4222")
publisher = EventPublisher(connection)

cloud_event = EventFactory.create(thermal_alarm)
published = await publisher.publish(cloud_event, "eventhub.raceway.process.alarm.thermalalarm")
# published.event_id, published.stream_sequence, published.duplicate
```

The application opens the NATS connection itself, with nats-py.

`publish` takes a CloudEvent built with `EventFactory`, then:

- checks the subject (`EventSubject.validate`);
- publishes the CloudEvent as structured JSON, with the `Content-Type: application/cloudevents+json` header and the
  `Nats-Msg-Id` header set to the event `id`: JetStream ignores the same event published again within its duplicate
  window, and `duplicate` is then `True`;
- waits for the JetStream acknowledgement. When no stream stores the subject, or JetStream refuses the event, it
  raises an `EventPublishError`. Connection errors (a closed connection…) come as the nats-py errors.

To publish an event again after an error, publish **the same CloudEvent**: it keeps its `id`, so JetStream stores it
only once. Building it again from the `EventModel` would give it a new `id`, and the event would be stored twice.

## Subscribing

```python
import asyncio
import logging

from industream.eventhub.client import ConsumeOptions, EventSubscriber, HandlerFailure, ReceivedEvent

subscriber = EventSubscriber(connection)


async def handle(received_event: ReceivedEvent) -> None:
    # react to received_event (decoded by EventDecoder)
    print(received_event.id)


def report(failure: HandlerFailure) -> None:
    logging.error("Event %s failed", failure.event.id, exc_info=failure.exception)


consumption = asyncio.create_task(
    subscriber.consume(
        "recipemaker-analyses",
        "eventhub.materialcatalog.>",
        handle,
        ConsumeOptions(on_handler_failure=report),
    )
)
# consumption.cancel() ends the consumption
```

The consumer name may not be empty nor contain whitespace, `.`, `*`, `>`, `/` or `\`, and `max_deliveries` must be
at least 1: `consume` raises a `ValueError` otherwise. `consume` runs until its task is cancelled, even before it has
started: it then hands no more event (those already received come back next time), and the handler is cancelled at
its next `await`. A cancelled handler is not a failure of the event. It uses a **durable consumer**: NATS remembers
where the consumer named `consumer_name` stopped, and the next `consume` with the same name resumes there, so an
application that was stopped receives the events it missed.

- An event is acknowledged once the handler succeeds. When the handler raises, the failure is passed to
  `on_handler_failure` and the event comes back a second later, up to `max_deliveries` deliveries (5 by default).
  After the last one (`is_last_attempt`), the consumer gives the event up; it stays in the stream.
- Events are handed one at a time, and delivered **at least once**: an event comes back when the application stops
  before acknowledging it, or when its handler runs longer than the acknowledgement wait of NATS (30 s). A handler
  must accept an event it has already handled (the event `id` tells).
- A message that is not an EventHub event is passed to `on_invalid_message` (an `InvalidMessage`: its NATS subject,
  stream sequence, body and error) and set aside for good, so that it does not come back in a loop.
- `start` only matters the first time the consumer is created: `SubscriptionStart.NEW_EVENTS` (default) receives the
  events published from then on, `SubscriptionStart.ALL_STORED_EVENTS` also receives the events still stored. A later
  `consume` with the same name keeps the position of the consumer, and updates its filter and `max_deliveries`.
- A callback that raises (`on_invalid_message`, `on_handler_failure`) ends `consume` with its exception, once the
  message is set aside or asked for again.

To read the past (replay), `read_from` returns, in order, the events stored from a start, then stops once caught up.
`follow_from` reads the same way, then goes on with the new events as they are published, until its task is
cancelled:

```python
async for received_event in subscriber.read_from(ReadStart.at(since), "eventhub.raceway.>"):
    ...

async for received_event in subscriber.follow_from(ReadStart.after(last_sequence), "eventhub.raceway.>"):
    last_sequence = received_event.stream_sequence
```

- The start is a date with a time zone (`ReadStart.at`) or the stream sequence of an event already read
  (`ReadStart.after`): every received event carries its position in the stream, `stream_sequence`. NATS cannot look
  an event up by its `id`.
- Unlike `consume`, they leave nothing behind in NATS: to resume later, the application keeps the `stream_sequence`
  of the last event it handled. A loop left before its end (a `break`) releases its NATS consumer when the generator
  is closed, which Python does later: wrap the generator in `contextlib.aclosing` to release it at once. Either way,
  NATS deletes a consumer nobody reads after a minute.
- A read survives the loss of its NATS consumer: NATS deletes it when the application takes more than a minute on the
  events fetched ahead (up to 256), or on a restart. The read then goes on right after the last event read, each event
  once and in order.
- They skip messages that are not EventHub events, after passing them to `ReadOptions.on_invalid_message`.
- When `ReadStart.after` asks for events already removed by the stream retention, `ReadOptions.on_missed_events` gets
  the requested and the oldest stored sequences before the read goes on from the oldest stored event. A start at a date
  cannot tell: NATS does not keep the time of the removed events.

All three check the filter with `EventSubject.validate_filter`: wildcards as whole tokens, `>` only last. Like
`EventSubject.validate_filter`, they name it `subject_filter`, since `filter` is a Python built-in.

## Modifying an event

Only the EventHub backend modifies an event: `EventModifier` sends it the **commands** (NATS Request-Reply, see the
backend README) and returns the event with its full new state.

```python
from industream.eventhub.client import CommandActor, EventModifier

modifier = EventModifier(connection)
actor = CommandActor.application("raceway")

acknowledged = await modifier.acknowledge(received_event.id, actor, "Coal injection reduced.")
ended = await modifier.end(received_event.id, actor)
# ended.state.acknowledgement, ended.state.end: actor (type, id, proof), time, comment
```

- `acknowledge` records that the actor has taken the event into account: any event, once, with an optional comment.
  `end` records that an event that lasts (`lasting`, an alarm) is over: once. They are independent: an acknowledged
  alarm stays active until it is ended.
- The actor decides the subject and the headers of the command, and the backend records it with the modification:

  | Actor | Who | Subject |
  |---|---|---|
  | `CommandActor.user(hub_token)` | A user, proven by their Industream Hub access token (`Authorization` header) | `eventhubcmd.<command>.user` |
  | `CommandActor.declared_user(user_name)` | A user named without proof (`Eventhub-Actor` header), for a backend whose authentication is off | `eventhubcmd.<command>.user` |
  | `CommandActor.application(application_name)` | An application, proven by the NATS permissions of its NATS user | `eventhubcmd.<command>.application.<name>` |
  | `CommandActor.flow(flow_name)` | A FlowMaker flow (`Eventhub-Actor` header), sent by the FlowMaker box | `eventhubcmd.<command>.flow` |

  An empty token, a blank name, a name with a control character, or an application name that is not a single
  subject token (whitespace, `.`, `*`, `>`) raises a `ValueError`, as does an empty event id.
- When the backend refuses the command, an `EventModificationError` carries its `status` (the HTTP meaning: 400
  instantaneous event or invalid request, 401 Hub token missing or invalid, 404 unknown event, 409 already acknowledged
  or ended), `title` and `detail`. When no backend answers, or not within 10 seconds, or its reply cannot be read,
  `status` is `None`. After a timeout, the modification may have been recorded: sent again, it gets 409.

## Reading the history

The stream keeps the events for its retention only (30 days by default), and NATS filters them by subject and date
only. The backend keeps every event in its database: `EventHistoryReader` reads them through its read queries (NATS
Request-Reply, see the backend README), with the state of each event.

```python
from datetime import UTC, datetime, timedelta

from industream.eventhub.client import EventHistoryFilter, EventHistoryReader, EventSeverity

reader = EventHistoryReader(connection)
history_filter = EventHistoryFilter(
    from_=datetime.now(UTC) - timedelta(days=7),
    projects=["raceway"],
    severities=[EventSeverity.ALARM],
    in_progress=True,
    attributes={"tuyere": 12},
)

page = await reader.read_history(history_filter, offset=0, limit=50)
# page.total; page.events: LatestEventVersion (event, nats_subject, state), most recent first
```

- Every filter is optional: the period `[from_, to[` (times with a time zone), projects, severities, categories,
  `acknowledged`, `in_progress` (events that last and are not ended), publisher attributes (equality) and `text`
  (contained in the type, the CloudEvents subject or an attribute value, whatever the case).
- The caller chooses the page: `offset` and `limit` (50 by default, 500 at most). `total` counts the events of the
  filter, all pages included.
- Only the main events are listed, each with the full state its modifications gave it (`EventState`, no acknowledgement
  and no end when never modified).
- When the backend refuses the query (400: an invalid filter or page), an `EventQueryError` carries its `status`,
  `title` and `detail`; when no backend answers, or not within 10 seconds, or its reply cannot be read, `status` is
  `None`. The queries are open: every NATS user that reads the stream may send them, without Hub token.

## Differences with the .NET client

What Python imposes, everything else being a translation:

- `EventSeverity` and `EventCategory` are enums with `UPPER_CASE` members, the Python convention:
  `EventSeverity.ALARM` is `EventSeverity.Alarm` in .NET. The value of an `EventSeverity` is its OpenTelemetry number,
  as in .NET; the value of an `EventCategory` is the name used by the contract (`"Process"`).
- An invalid value raises a `ValueError`, the Python counterpart of `ArgumentException`.
- `time` is a `datetime` that must have a time zone: a naive `datetime` would be read in the local time of the
  machine, a silent error.
- A publisher attribute is a `str`, a 32-bit `int`, a `bool` or a `datetime`: Python has no URI type, so a URI is
  given as a `str`.
- `CloudEvent` is a type of this client, a read-only mapping of the structured JSON, not the class of the CloudEvents
  SDK as in .NET. The Python SDK (`cloudevents` 2.x) checks almost none of the rules of the contract: it accepts any
  extension value, reads a time at `24:00` or written with a lowercase `t` or `z`, and writes the time to the
  microsecond (`…42.318000Z`) where the contract wants `…42.318Z`. Every rule would then have to be written around it
  anyway, in two places and with two more dependencies (`cloudevents`, `python-dateutil`). Instead, `EventFactory`,
  `EventJson` and `EventDecoder` follow the CloudEvents rules themselves, held by the shared contract cases, as the
  TypeScript client does.
- `ReceivedEvent` gives `None` for a missing value, `root_event_id` as a `UUID` (the `Guid` of .NET), `time` as a
  `datetime` in UTC (to the microsecond, the precision of `datetime`), and `data` as parsed JSON.
- `EventDecoder` checks what the .NET SDK checks: the required attributes (`id`, `source`, `type`, `specversion`
  1.0, and `time` for EventHub), the types of the standard attributes (an RFC 3339 `time`, a URI reference
  `source`…), and extension values of the CloudEvents type system only. It also rejects `NaN` and `Infinity`, which
  the `json` module reads although they are not JSON.
- The client is asynchronous (asyncio), as nats-py is. Cancelling the task replaces the .NET `CancellationToken`, the
  asyncio convention: `consume`, `read_from` and `follow_from` take no token, and the handler only gets the event.
  A cancelled `consume` or `follow_from` ends with `asyncio.CancelledError`, as any cancelled coroutine, where .NET
  returns quietly; nothing else is raised, and a cancelled handler is not reported as a failure. To stop an
  application, cancel these tasks first, then close or drain the connection: once the connection is closed, nats-py
  ends what still waits for messages with a `ConnectionClosedError`.
- The error messages are those of .NET, in the Python words of the TypeScript client where the types differ
  (`max_deliveries`, `None`, `int`): the shared contract cases check the same parts of them.
- `publish` takes no cancellation either: cancelling its task stops the wait for the acknowledgement. Nor do
  `acknowledge` and `end`: cancelling their task stops the wait for the reply.
- `EventHistoryFilter.from_` is the `From` of .NET: `from` is a keyword of Python. A time of the filter without time
  zone raises a `ValueError`, as `ReadStart.at` does.
- `ActorType` and `ActorProof` are enums with `UPPER_CASE` members (`ActorProof.HUB_TOKEN`), valued with the .NET
  names (`"HubToken"`); the times of `EventState` are `datetime`s in UTC.
- `HandlerFailure.exception` is the `Exception` the handler raised; the callbacks of `ConsumeOptions` and
  `ReadOptions` are plain functions, as the `Action` callbacks of .NET.
- `follow_from` reads with an ordered consumer of this client, where .NET uses the one of NATS.Net: an ephemeral pull
  consumer kept in memory, created again right after the last event read when it is lost (a NATS restart, a
  disconnection or a reader slower than a minute) or when a message is missing. `read_from` reads with the same one,
  stopping once caught up, as .NET recreates its read consumer. The ordered consumer of nats-py is not used: when
  it creates its consumer again, it keeps the start date next to the new start sequence, which the server refuses
  (`follow_from` would then wait for good), and it starts from the beginning of the stream when no event was read yet.

## Build and test

From this folder, with [uv](https://docs.astral.sh/uv/):

```bash
uv sync
uv run ruff check
uv run ruff format --check
uv run mypy
uv run pytest
```

`uv run pytest tests/unit` runs the unit tests only; `uv run pytest tests/e2e` runs the end-to-end tests, which start a
throwaway NATS server with Testcontainers: Docker must be running.
