python-event-sourcery · lzukowski · Mar 17, 2026 · Mar 14, 2026 · Mar 15, 2026 · Mar 14, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.5.2
+### Changed
+- **Breaking:** `Repository.aggregate()` yields `WrappedAggregate` instead of `Aggregate` directly
+
 ## 0.5.1
 ### Added
 - In transaction subscription to category

diff --git a/docs/adr/20260314-aggregate-metadata.md b/docs/adr/20260314-aggregate-metadata.md
@@ -0,0 +1,143 @@
+# Aggregate Metadata
+
+Date: 2026-03-14
+
+## Status
+
+Accepted
+
+## Context
+
+Currently `Repository.aggregate()` yields only the aggregate instance. The caller has no access to stream-level information such as:
+
+- What is the current version of the aggregate?
+- Is this a new aggregate (no prior events in the stream)?
+- When was the stream created or last modified?
+- What stream identity was used?
+
+This information is essential for many use cases:
+- Detecting whether an aggregate is being created for the first time (e.g. to enforce idempotent creation)
+- Knowing the version for external optimistic concurrency checks or ETags
+- Accessing timestamps for auditing or display purposes
+
+Other event sourcing frameworks consistently expose this kind of metadata:
+- **Marten** returns `StreamState` with `Version`, `Created`, `LastTimestamp`, `AggregateType` via `FetchForWriting<T>`
+- **Eventide** returns `(entity, version)` tuple, with `:no_stream` sentinel for non-existent streams
+- **Axon** exposes `@AggregateVersion` and `AggregateLifecycle.isLive()`
+- **Ecotone** enriches events with `_aggregate_version`, `_aggregate_type` headers
+
+## Decisions
+
+### D1: Wrapper pattern — `WrappedAggregate`
+
+Following the established pattern of `WrappedEvent` which wraps `Event` with metadata, the repository will return a `WrappedAggregate` that wraps the aggregate instance with stream metadata. The aggregate itself remains a pure domain object with no knowledge of infrastructure concerns.
+
+```python
+with repo.aggregate(uuid, LightSwitch()) as wrapped:
+    if wrapped.is_new:
+        wrapped.aggregate.turn_on()
+```
+
+**Rationale:** This is consistent with `WrappedEvent[TEvent]` which carries `event`, `version`, `uuid`, `created_at`, and `context` alongside the event itself. `WrappedAggregate[TAggregate]` follows the same convention — the wrapper holds metadata, the inner object stays pure.
+
+**Migration:** This is a breaking change. Existing callers using `as aggregate` will need to change to `as wrapped` and access `wrapped.aggregate`. Given the library is pre-1.0, this is acceptable.
+
+### D2: Fields on `WrappedAggregate`
+
+| Field | Type | Source |
+|---|---|---|
+| `aggregate` | `TAggregate` | The aggregate instance |
+| `version` | `int` (computed) | Stored version + count of pending changes in aggregate |
+| `is_new` | `bool` (computed) | `True` when stored version is 0 (no events existed before load) |
+| `stream_id` | `StreamId` | Built from UUID + aggregate category |
+| `created_at` | `datetime \| None` | Timestamp of the first event in the stream |
+| `updated_at` | `datetime \| None` | Timestamp of the last event in the stream |
+| `context` | `Context \| None` | The context passed to the repository for this operation |
+
+**`version` is computed dynamically.** It reflects the stored version plus any uncommitted changes the aggregate has emitted. This means:
+- After loading an aggregate with 5 events: `version == 5`
+- After emitting 2 more events: `version == 7`
+- After persisting (context exit): version reflects the final stored state
+
+This is achieved by reading the aggregate's `__changes__` property and computing `stored_version + len(aggregate.__changes__)` on access.
+
+**`is_new` is based on stored version only**, not including pending changes. An aggregate that was just created and has emitted its first event is still `is_new == True` — it didn't exist before this session. This matches Eventide's `:no_stream` semantics.
+
+### D3: Class naming and location
+
+The class will be called `WrappedAggregate`. It is a generic dataclass `WrappedAggregate[TAggregate]`, mirroring `WrappedEvent[TEvent]`. It lives in the `event_sourcery.event_sourcing` module alongside `Aggregate` and `Repository`.
+
+**Rationale:** `WrappedAggregate` directly mirrors `WrappedEvent` — both are generic wrappers that pair a domain object with infrastructure metadata. The naming convention is already established in the project. A dataclass is used instead of a Pydantic model because `WrappedAggregate` is a simple data holder — it does not need serialization, validation, or schema generation. A dataclass keeps the dependency footprint minimal and the implementation straightforward.
+
+## Solution Proposal
+
+### New class: `WrappedAggregate`
+
+```python
+# event_sourcery/event_sourcing/aggregate.py
+
+TAggregate = TypeVar("TAggregate", bound=Aggregate)
+
+@dataclass
+class WrappedAggregate(Generic[TAggregate]):
+    aggregate: TAggregate
+    stream_id: StreamId
+    context: Context | None
+    created_at: datetime | None
+    updated_at: datetime | None
+    stored_version: int
+
+    @property
+    def version(self) -> int:
+        return self.stored_version + len(
+            getattr(self.aggregate, "__changes__", [])
+        )
+
+    @property
+    def is_new(self) -> bool:
+        return self.stored_version == 0
+
+```
+
+### Changes to `Repository`
+
+```python
+# event_sourcery/event_sourcing/repository.py
+
+@contextmanager
+def aggregate(
+    self,
+    uuid: StreamUUID,
+    aggregate: TAggregate,
+    context: Context | None = None,
+) -> Iterator[WrappedAggregate[TAggregate]]:
+    stream_id = StreamId(uuid=uuid, name=uuid.name, category=aggregate.category)
+    stored_version, created_at, updated_at = self._load(stream_id, aggregate)
+    wrapped = WrappedAggregate(
+        aggregate=aggregate,
+        stream_id=stream_id,
+        context=context,
+        created_at=created_at,
+        updated_at=updated_at,
+        stored_version=stored_version,
+    )
+    yield wrapped
+    self._save(aggregate, stored_version, stream_id, context)
+```
+
+The `_load` method will be updated to extract `created_at` from the first event's timestamp and `updated_at` from the last event's timestamp during replay.
+
+### Public API exports
+
+`WrappedAggregate` will be exported from `event_sourcery.event_sourcing` package `__init__.py`.
+
+## Consequences
+
+- **Breaking change** in `Repository.aggregate()` yield type — all callers must update from `as aggregate` to `as wrapped` and access `wrapped.aggregate`.
+- Consistent with the existing `WrappedEvent` pattern — both domain objects (`Event`, `Aggregate`) are wrapped with metadata by infrastructure, never polluted directly.
+- The aggregate remains a pure domain object — no infrastructure leakage.
+- `version` property provides a live view of the aggregate's version including pending changes, useful for logging/debugging.
+- `is_new` enables idempotent aggregate creation patterns without checking version manually.
+- `created_at` / `updated_at` are derived from event timestamps — no additional storage or queries needed.
+- `context` reference enables callers to inspect or pass along the context used for the current operation.
+- `version` reads `aggregate.__changes__` dynamically, so it updates automatically as events are emitted.
diff --git a/docs/code/test_recipes.py b/docs/code/test_recipes.py
@@ -289,9 +289,9 @@ def switch_on(self) -> None:
     from event_sourcery import StreamUUID
 
     stream_id = StreamUUID(name="light_switch/1")
-    with repository.aggregate(stream_id, LightSwitch()) as light_switch:
-        light_switch.switch_on()
-        light_switch.switch_on()
+    with repository.aggregate(stream_id, LightSwitch()) as wrapped:
+        wrapped.aggregate.switch_on()
+        wrapped.aggregate.switch_on()
     # --8<-- [end:event_sourcing_03]
 
     events = backend.event_store.load_stream(

diff --git a/docs/concepts/basics.md b/docs/concepts/basics.md
@@ -188,71 +188,84 @@ To sum up, [Event Sourcing] comes down to:
 Event Sourcery provides a base class for an [Aggregate] and repository implementation that makes it much easy to create or read/change [Aggregate].
 
 ```python
+from event_sourcery.event import Event
+from event_sourcery.event_sourcing import Aggregate
+
+
+class TurnedOn(Event):
+    pass
+
+
+class TurnedOff(Event):
+    pass
+
+
 class LightSwitch(Aggregate):
     """A simple aggregate that models a light switch."""
+    category = "light_switch"  # stream category for all LightSwitch aggregates
+
     class AlreadyTurnedOn(Exception):
         pass
 
     class AlreadyTurnedOff(Exception):
         pass
 
-    def __init__(
-        self, past_events: list[Event], changes: list[Event], stream_id: StreamId
-    ) -> None:
+    def __init__(self) -> None:
         # init any state you need in aggregate class to check conditions
         self._shines = False
 
-        # required for base class
-        super().__init__(past_events, changes, stream_id)
-
-    def _apply(self, event: Event) -> None:
-        # each aggregate need an _apply method to parse events
+    def __apply__(self, event: Event) -> None:
+        # each aggregate needs an __apply__ method to handle events
         match event:
-            case TurnedOn() as event:
+            case TurnedOn():
                 self._shines = True
-            case TurnedOff() as event:
+            case TurnedOff():
                 self._shines = False
 
     def turn_on(self) -> None:
-        # this is one of command methods
-        # we can rejest it (i.e. raise an exception)
+        # this is a command method
+        # we can reject it (i.e. raise an exception)
         # if current state does not allow this to proceed
-        # e.g. light is already on
         if self._shines:
             raise LightSwitch.AlreadyTurnedOn
-        self._event(TurnedOn)
+        self._emit(TurnedOn())
 
     def turn_off(self) -> None:
         if not self._shines:
             raise LightSwitch.AlreadyTurnedOff
-
-        self._event(TurnedOff)
+        self._emit(TurnedOff())
 ```
 
-To create a [Repository] tailored for a particular [Aggregate] class, we need  that class and [EventStore] instance: 
+To create a [Repository] tailored for a particular [Aggregate] class, we need an [EventStore] instance:
 
 ```python
-repository = Repository[LightSwitch](event_store, LightSwitch)
+from event_sourcery.event_sourcing import Repository
+
+repository = Repository[LightSwitch](event_store)
 ```
 
-A [Repository] exposes method to create a new instance of [Aggregate]:
+A [Repository] exposes a context manager to work with an [Aggregate]. It returns a [WrappedAggregate] that wraps the aggregate with stream metadata (version, timestamps, etc.):
 
 ```python
-stream_id = uuid4()
+from event_sourcery import StreamUUID
+
+uuid = StreamUUID()
 
-with repository.new(stream_id=stream_id) as switch:
-    switch.turn_on()
+with repository.aggregate(uuid, LightSwitch()) as wrapped:
+    wrapped.aggregate.turn_on()
+    print(wrapped.version)   # 1
+    print(wrapped.is_new)    # True
 ```
 
-...or to work with existing [Aggregate], making sure changes are saved at the end:
+Loading an existing [Aggregate] works the same way — events are replayed automatically:
 
 ```python
-with repository.aggregate(stream_id=stream_id) as switch_second_incarnation:
+with repository.aggregate(uuid, LightSwitch()) as wrapped:
     try:
-        switch_second_incarnation.turn_on()
+        wrapped.aggregate.turn_on()
     except LightSwitch.AlreadyTurnedOn:
         # o mon Dieu, I made a mistake!
-        switch_second_incarnation.turn_off()
+        wrapped.aggregate.turn_off()
 ```
 
 A [Repository] is a thin wrapper over Event Store. One can also write Aggregates even without using our base class and use [EventStore] directly!
@@ -262,3 +275,4 @@ A [Repository] is a thin wrapper over Event Store. One can also write Aggregates
 [Event Sourcing]: ../recipes/event_sourcing.md
 [Aggregate]: ../reference/event_sourcing/Aggregate.md
 [Repository]: ../reference/event_sourcing/Repository.md
+[WrappedAggregate]: ../reference/event_sourcing/WrappedAggregate.md
diff --git a/docs/recipes/event_sourcing.md b/docs/recipes/event_sourcing.md
@@ -2,7 +2,7 @@
 
 Event Sourcery provides a few building blocks to work with event sourcing.
 
-These are [Aggregate] and [Repository] base classes.
+These are [Aggregate], [Repository] and [WrappedAggregate] classes.
 
 ## Usage
 
@@ -12,7 +12,7 @@ There are three required attributes that need to be defined:
 
 1. `category` class-level constant that will be added to all streams from all aggregates of this type
 2.  `__init__` if defined, must not accept any arguments
-3. `__apply__` method that will change internal state of the aggregate based on the event applied during reading state from the event store 
+3. `__apply__` method that will change internal state of the aggregate based on the event applied during reading state from the event store
 
 ```python
 --8<--
@@ -28,14 +28,22 @@ docs/code/test_recipes.py:event_sourcing_02_repo
 --8<--
 ```
 
-From now on, regardless if you want to work with a given aggregate for the first time or load existing one, you should use `repository.aggregate` context manager:
+From now on, regardless if you want to work with a given aggregate for the first time or load existing one, you should use `repository.aggregate` context manager. It returns a [WrappedAggregate] — a wrapper that provides the aggregate instance along with stream metadata such as `version`, `is_new`, `created_at`, and `updated_at`:
 
 ```python
 --8<--
 docs/code/test_recipes.py:event_sourcing_03
 --8<--
 ```
 
+The aggregate itself is accessed via `wrapped.aggregate`. The wrapper also exposes useful properties:
+
+- `wrapped.version` — current version including pending (not yet persisted) changes
+- `wrapped.is_new` — `True` if no events existed before this session
+- `wrapped.created_at` / `wrapped.updated_at` — timestamps of first and last event in the stream
+- `wrapped.stream_id` — the stream identity (UUID + category)
+
 [Aggregate]: ../reference/event_sourcing/Aggregate.md
 [Repository]: ../reference/event_sourcing/Repository.md
+[WrappedAggregate]: ../reference/event_sourcing/WrappedAggregate.md
 [EventStore]: ../reference/event_store/EventStore.md
diff --git a/docs/reference/event_sourcing/WrappedAggregate.md b/docs/reference/event_sourcing/WrappedAggregate.md
@@ -0,0 +1 @@
+::: event_sourcery.event_sourcing.WrappedAggregate
diff --git a/event_sourcery/event_sourcing/__init__.py b/event_sourcery/event_sourcing/__init__.py
@@ -1,7 +1,8 @@
 __all__ = [
     "Aggregate",
     "Repository",
+    "WrappedAggregate",
 ]
 
-from event_sourcery.event_sourcing.aggregate import Aggregate
+from event_sourcery.event_sourcing.aggregate import Aggregate, WrappedAggregate
 from event_sourcery.event_sourcing.repository import Repository
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		::: event_sourcery.event_sourcing.WrappedAggregate