Metadata-Version: 2.5
Name: industream-eventhub-client
Version: 1.0.0rc1
Summary: EventHub client for Python: publishes and subscribes to Industream events stored in NATS JetStream.
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.

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);
- 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), 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`; 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` 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.

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

## 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.
- `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 longer than a minute) or when a message is missing. 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.
