WiP

depeche_db/_subscription.py

@@ -223,6 +223,8 @@ def get_next_messages(self, count: int) -> Iterator[SubscriptionMessage[E]]:
                     count=count,
                 ):
                     if isinstance(message, DeletedAggregatedStreamMessage):
+                        # TODO needs to update the state so that we can continue
+                        # if there are more deleted messages in a row than `count`
                         continue
                     ack = AckOp(
                         name=self.name,

examples/basic.py

@@ -6,13 +6,12 @@
 import sqlalchemy as _sa
 
 from depeche_db import (
-    MessageProtocol,
     MessageStore,
 )
 from depeche_db.tools import PydanticMessageSerializer
 
 
-class MyEvent(_pydantic.BaseModel, MessageProtocol):
+class MyEvent(_pydantic.BaseModel):
     event_id: _uuid.UUID = _pydantic.Field(default_factory=_uuid.uuid4)
     num: int
     happened_at: _dt.datetime = _pydantic.Field(default_factory=_dt.datetime.utcnow)
@@ -43,7 +42,7 @@ class EventB(MyEvent):
 )
 
 stream = f"stream-{_uuid.uuid4()}"
-events = [EventA(num=n + 1) for n in range(3)]
+events = [EventA(num=n + 1) for n in range(4)]
 
 result = message_store.write(stream=stream, message=events[0])
 print(result)
@@ -65,3 +64,7 @@ class EventB(MyEvent):
 message_store.synchronize(stream=stream, messages=events, expected_version=3)
 print("Now we have 4 events")
 print([e.message.num for e in message_store.read(stream)])
+
+message_store.delete(stream=stream, keep_versions_greater_than=-2)
+print("Now we have only the last 2 events")
+print([e.message.num for e in message_store.read(stream)])

tests/aggregated_stream/test_delete.py

@@ -18,3 +18,14 @@ def test_deleted_events_in_stream(db_engine, store_with_events, stream_factory):
         ),
     ]
     assert type(messages[-1]) == LoadedAggregatedStreamMessage
+
+
+def test_case():
+    # TODO:
+    #  * Add 1 message to a stream
+    #  * set up aggregated stream and update it
+    #  * add 2 more messages
+    #  * delete all but the last message
+    #  * update aggregated stream
+    # -> what happened?!
+    pass