Skip to content

StreamId

Bases: StreamUUID

Identifier for an event stream, with optional category support.

Extends StreamUUID by adding a category attribute, allowing grouping of streams (e.g., by aggregate type or business domain). Used as the primary identifier for event streams in the event store, supporting both UUID-based and name-based addressing.

Attributes:

Name Type Description
category StreamCategory | None

Optional category for grouping streams.
Source code in event_sourcery/_event_store/stream_id.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@dataclass(frozen=True, repr=False, eq=False)
class StreamId(StreamUUID):
    """
    Identifier for an event stream, with optional category support.

    Extends StreamUUID by adding a category attribute, allowing grouping of streams
    (e.g., by aggregate type or business domain). Used as the primary identifier for
    event streams in the event store, supporting both UUID-based and name-based
    addressing.

    Attributes:
        category (StreamCategory | None): Optional category for grouping streams.
    """

    category: StreamCategory | None = None

    def __repr__(self) -> str:
        return (
            f"{type(self).__name__}"
            f"(hex={self!s}, name={self.name}, category={self.category})"
        )

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, StreamId):
            return super().__eq__(other) and self.category == other.category
        return NotImplemented

    def __hash__(self) -> int:
        return hash((self.category, super().__hash__()))