Skip to content

castlecraft_engineer.abstractions.projection

castlecraft_engineer.abstractions.projection

ProjectionState dataclass

Represents the state of a projection, typically used to track its progress in processing an event stream.

Source code in src/castlecraft_engineer/abstractions/projection.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@dataclass
class ProjectionState:
    """
    Represents the state of a projection, typically used to track
    its progress in processing an event stream.
    """

    projection_id: ProjectionId
    # Could be UUID, int, str
    last_processed_event_id: Optional[Any] = None
    last_processed_event_timestamp: Optional[datetime.datetime] = None
    last_updated_at: datetime.datetime = field(
        default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
    )

    def update_progress(
        self, event_id: Any, event_timestamp: Optional[datetime.datetime]
    ):
        self.last_processed_event_id = event_id
        if event_timestamp:
            self.last_processed_event_timestamp = event_timestamp
        self.last_updated_at = datetime.datetime.now(datetime.timezone.utc)