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 and 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

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.

Building an event

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

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.

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.

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.

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

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:

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

Publishing an event

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

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:

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.

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.

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.

Statistics

read_statistics reads the numbers of the main events of a period, as the Overview of the UI shows them:

to = datetime.now(UTC)
statistics = await reader.read_statistics(to - timedelta(hours=24), to, timedelta(hours=1), "raceway", 10)
# statistics.by_severity, total: of the project (or of every project when None)
# statistics.by_project: every project of the period (by_severity, total, open_alarms)
# statistics.top_types: type, project, severity, count; most frequent first
# statistics.histogram: start, end, by_severity; every bucket of the period

• The caller chooses the width of the histogram buckets, in whole seconds: the backend refuses more than 500 buckets. The last bucket ends at the end of the period.
• open_alarms counts the alarms of the period that last and are not ended yet. The numbers per severity give every severity, 0 included.
• Same errors as the history: an EventQueryError.

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 datetimes 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:

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.