simple¶
Implements SimpleCheckpoint, a JSON-file-based checkpoint for local
development and testing.
- class unistream.checkpoints.simple.SimpleCheckpoint(lock_expire: int = REQ, max_attempts: int = REQ, initial_pointer: str | int = REQ, start_pointer: str | int = REQ, next_pointer: str | int | None = REQ, batch_sequence: int = REQ, batch: dict[str, Tracker] = REQ, checkpoint_file: str = REQ, records_file: str = REQ)[source]¶
A simple checkpoint using local json file for persistence.
- Parameters:
path_checkpoint – the path to the checkpoint file.
path_records – the path to the records data file.
- classmethod load(checkpoint_file: str, records_file: str, lock_expire: int = 900, max_attempts: int = 3, initial_pointer: str | int = 0, start_pointer: str | int = 0, next_pointer: str | int | None = None, batch_sequence: int = 0, batch: dict[str, Tracker] | None = None) SimpleCheckpoint[source]¶
Load the checkpoint data from the persistence layer.
It has to handle the edge case that the checkpoint data does not exist.
- dump_records(records: Iterable[AbcRecord])[source]¶
Dump the records in a batch to the persistence layer.
- load_records(record_class: type[AbcRecord], **kwargs) list[AbcRecord][source]¶
Load the batch records from the persistence layer.
- dump_as_in_progress(record: AbcRecord)[source]¶
Dump the tracker to the persistence layer after calling
BaseCheckpoint.mark_as_in_progress.Note
It is up to the developer to implement the persistence layer to persist the data.
- Parameters:
record – the record to check.