@@ -0,0 +1,1857 @@
+%22%22%22:mod:%60kinsumer.checkpointer%60 --- Persisting positions for Kinesis shards%0A~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~%0A%0A%22%22%22%0Aimport abc%0Aimport json%0Aimport os.path%0Afrom typing import Optional, Dict%0A%0A%0Aclass Checkpointer(abc.ABC, object):%0A    %22%22%22Checkpointer is the interface for persisting positions for Kinesis shards%0A%0A    %22%22%22%0A%0A    @abc.abstractmethod%0A    def get_checkpoints(self) -%3E Dict%5Bstr, str%5D:%0A        %22%22%22Get a dictionary whose keys are all the shard ids we are aware of,%0A        and whose values are the sequence id of the last record processed for%0A        its shard%0A%0A        %22%22%22%0A%0A    @abc.abstractmethod%0A    def checkpoint(self, shard_id: str, sequence: str) -%3E None:%0A        %22%22%22Persist the sequence number for a given shard%22%22%22%0A%0A    @abc.abstractmethod%0A    def get_checkpoint(self, shard_id: str) -%3E Optional%5Bstr%5D:%0A        %22%22%22Get the sequence number of the last successfully processed record%22%22%22%0A%0A%0Aclass InMemoryCheckpointer(Checkpointer):%0A    def __init__(self) -%3E None:%0A        self._checkpoints = %7B%7D%0A%0A    def get_checkpoints(self) -%3E Dict%5Bstr, str%5D:%0A        return self._checkpoints.copy()%0A%0A    def checkpoint(self, shard_id: str, sequence: str) -%3E None:%0A        self._checkpoints%5Bshard_id%5D = sequence%0A%0A    def get_checkpoint(self, shard_id: str) -%3E Optional%5Bstr%5D:%0A        return self._checkpoints.get(shard_id)%0A%0A%0Aclass FileCheckpointer(InMemoryCheckpointer):%0A    def __init__(self, file: str) -%3E None:%0A        super().__init__()%0A        self.file = file%0A        if os.path.exists(file):%0A            with open(self.file, 'rb') as f:%0A                self._checkpoints = json.load(f)%0A%0A    def checkpoint(self, shard_id: str, sequence: str) -%3E None:%0A        super().checkpoint(shard_id, sequence)%0A        with open(self.file, 'wb') as f:%0A            f.write(json.dumps(self._checkpoints, ensure_ascii=False).encode())%0A

Add checkpointer