from dataclasses import dataclass, field from io import BufferedRandom from os.path import getsize from time import time @dataclass class Shard: size: int file: BufferedRandom file_size: int @staticmethod def load_shard(filename: str) -> 'Shard': file = open(filename, 'r+b') size = int.from_bytes(file.read(8), 'big') file.seek(0, 2) file_size = file.tell() return Shard(size, file, file_size) def write_entry(self, entry: 'Entry'): data = entry.to_bytes() data_length = len(data) if self.size + 1 + data_length > self.file_size: raise EOFError() self.file.seek(self.size + 1) self.file.write(data) self.size += data_length self.file.seek(0) self.file.write(self.size.to_bytes(8, 'big')) def close(self): self.file.close() @dataclass class Entry: shard: Shard timestamp_start: int = field(default_factory=lambda: int(time())) timestamp_end: int = field(default_factory=lambda: int(time())) content: bytes = field(default_factory=bytes) @staticmethod def from_bytes(shard: Shard, data: BufferedRandom) -> 'Entry': timestamp_start = int.from_bytes(data.read(8), 'big') timestamp_end = int.from_bytes(data.read(8), 'big') content_length = int.from_bytes(data.read(8), 'big') content = data.read(content_length) return Entry(shard, timestamp_start, timestamp_end, content) def to_bytes(self) -> bytes: data = self.timestamp_start.to_bytes(8, 'big') data += self.timestamp_end.to_bytes(8, 'big') data += len(self.content).to_bytes(8, 'big') data += self.content return data def add_content(self, content: bytes): self.timestamp_end = int(time()) self.content += content