Smithy AWS Event Stream

0.2.1 · active · verified Fri Apr 17

Smithy AWS Event Stream provides components for interacting with Amazon Event Streams. It handles the serialization and deserialization of event stream messages according to the AWS Event Stream specification. This library is part of the broader `smithy-python` project, currently at version 0.2.1, with a release cadence tied to the evolution of the `smithy-python` ecosystem.

Common errors

```
ERROR: Package 'smithy-aws-event-stream' requires a different Python: ...
```
cause Your current Python environment does not meet the minimum requirement of Python 3.12 for this library.

fix Upgrade your Python installation to version 3.12 or higher, or use a virtual environment configured with a compatible Python version.
```
AttributeError: module 'smithy_aws_event_stream' has no attribute 'EventStreamMessage'
```
cause You are trying to import `EventStreamMessage` (or similar data model classes) directly from the top-level package.

fix These classes are located in the `model` submodule. Use `from smithy_aws_event_stream.model import EventStreamMessage`.
```
TypeError: object async_generator is not iterable
```
cause You are attempting to iterate over an asynchronous generator or iterator (like `EventStreamSerde`) using a synchronous `for` loop.

fix Use the `async for` syntax within an `async` function to correctly iterate over asynchronous iterators: `async for message in async_iterator: ...`

Warnings

breaking As a pre-1.0 library (0.x.x versioning), API stability is not guaranteed. Expect breaking changes between minor versions (e.g., 0.1.x to 0.2.x) as the library evolves.
Fix: Always refer to the latest `CHANGELOG.md` in the `smithy-python` repository and pin your dependencies to specific versions to mitigate unexpected breakage.
gotcha This library requires Python 3.12 or higher. Attempting to install or run it on older Python versions will result in dependency resolution failures or runtime errors.
Fix: Ensure your development and deployment environments are running Python 3.12 or a newer compatible version.
gotcha The library is designed for asynchronous I/O and heavily uses async iterators/generators for handling event streams. Incorrect handling of asynchronous patterns (e.g., missing `await`, improper loop management) can lead to unexpected behavior, resource leaks, or deadlocks.
Fix: Familiarize yourself with Python's `asyncio` and `async`/`await` patterns. Always use `async for` when iterating over `EventStreamSerde` or similar async iterators, and properly `await` async functions.

Install

pip install smithy-aws-event-stream Install stable version

Imports

EventStreamMessage
wrong:
```
from smithy_aws_event_stream import EventStreamMessage
```
correct:
```
from smithy_aws_event_stream.model import EventStreamMessage
```
Core data model classes are nested under the `model` submodule, not directly at the top level.
EventStreamSerde
```
from smithy_aws_event_stream.serde import EventStreamSerde
```
The serializer/deserializer class is in the `serde` submodule and typically used within a `smithy-python` generated client.

Quickstart

This quickstart demonstrates how to construct a basic `EventStreamMessage` using the library's data models. This message can then be serialized and sent over an event stream, typically within the context of an AWS client generated by `smithy-python`.

from smithy_aws_event_stream.model import EventStreamMessage, EventStreamHeader, EventStreamHeaderValue

# 1. Create headers for the event stream message
headers = [
    EventStreamHeader(name=":message-type", value=EventStreamHeaderValue.string("event")),
    EventStreamHeader(name=":event-type", value=EventStreamHeaderValue.string("MyApplicationEvent")),
    EventStreamHeader(name="correlation-id", value=EventStreamHeaderValue.uuid("a1b2c3d4-e5f6-7890-1234-567890abcdef")),
    EventStreamHeader(name="is-urgent", value=EventStreamHeaderValue.boolean(True)),
    EventStreamHeader(name="sequence-number", value=EventStreamHeaderValue.int64(1))
]

# 2. Define the payload (can be empty or any bytes)
payload = b'{"data": "This is my event payload!"}'

# 3. Create an EventStreamMessage instance
message = EventStreamMessage(headers=headers, payload=payload)

print("Created Event Stream Message:")
print(f"  Payload: {message.payload.decode('utf-8')}")
print("  Headers:")
for header in message.headers:
    # Using to_python() to get the native Python value from EventStreamHeaderValue
    print(f"    - {header.name}: {header.value.to_python()}")

# In a real-world scenario, this message would typically be serialized
# by an EventStreamSerde instance (part of an AWS client generated by smithy-python)
# and sent over an asynchronous stream to an AWS service.

view raw JSON →