| 9 | self.diff = diff |
| 10 | |
| 11 | class State: |
| 12 | def __init__(self, collect_history: bool = False): |
| 13 | self.state = {} |
| 14 | self.timestamp = 0 |
| 15 | self.valid = True |
| 16 | self.history = [] if collect_history else None |
| 17 | |
| 18 | def get_state(self) -> List[T]: |
| 19 | result = [] |
| 20 | for key, value in self.state.items(): |
| 21 | clone = json.loads(key) |
| 22 | for i in range(int(value)): |
| 23 | result.append(clone) |
| 24 | return result |
| 25 | |
| 26 | def get_history(self): |
| 27 | return self.history |
| 28 | |
| 29 | def validate(self, timestamp: int): |
| 30 | if not self.valid: |
| 31 | raise Exception("Invalid state.") |
| 32 | elif timestamp < self.timestamp: |
| 33 | print("Invalid timestamp.") |
| 34 | self.valid = False |
| 35 | raise Exception(f"Update with timestamp ({timestamp}) is lower than the last timestamp ({self.timestamp}). Invalid state.") |
| 36 | |
| 37 | def process(self, update: Update): |
| 38 | value = json.dumps(update['value']) |
| 39 | count = self.state.get(value, 0) + update['diff'] |
| 40 | |
| 41 | if count <= 0: |
| 42 | del self.state[value] |
| 43 | else: |
| 44 | self.state[value] = count |
| 45 | |
| 46 | if self.history is not None: |
| 47 | self.history.append(update) |
| 48 | |
| 49 | def update(self, updates: List[Update], timestamp: int): |
| 50 | if len(updates) > 0: |
| 51 | self.validate(timestamp) |
| 52 | self.timestamp = timestamp |
| 53 | for update in updates: |
| 54 | self.process(update) |
no outgoing calls
no test coverage detected