Source code for graph.data.key_value_store

import typing


[docs] class KeyValueStore: """ In-memory key-value store that saves, retrieves, and merges values by type or custom key. Attributes: storage (dict[str, object]): Underlying dictionary holding stored values. """
[docs] def __init__(self): """ Initialize an empty KeyValueStore. """ self.storage: dict[str, object] = {}
def __contains__(self, key: type) -> bool: """ Check if a value of the given type is stored. Args: key (type): The class/type to check in storage. Returns: bool: True if a value of this type exists, False otherwise. Raises: ValueError: If `key` is not a type. """ if not isinstance(key, type): raise ValueError(f"Key must be a type, not {type(key)}") return key.__name__ in self.storage
[docs] def save(self, value: object) -> None: """ Store a value under its class name. Args: value (object): The object to store. Raises: ValueError: If a value of the same type is already stored. """ key = type(value).__name__ if key in self.storage: raise ValueError(f"Key {key} already exists in storage") self.storage[key] = value
[docs] def get(self, value_type: type) -> object: """ Retrieve a stored value by its class type. Args: value_type (type): The class/type of the value to retrieve. Returns: object: The stored object. Raises: KeyError: If no value of this type is found. """ return self.storage[value_type.__name__]
[docs] def save_by_key(self, key: str, value: object) -> None: """ Store a value under a custom string key. Args: key (str): The string key to use in storage. value (object): The object to store. """ self.storage[key] = value
[docs] def get_by_key(self, key: str) -> object: """ Retrieve a value by a custom string key. Args: key (str): The key to look up. Returns: object: The stored object. Raises: KeyError: If the key is not found. """ return self.storage[key]
[docs] def save_tickets(self, tickets: list[object]) -> None: """ Store a list of tickets under the 'tickets' key. Args: tickets (list[object]): List of ticket objects to store. """ self.storage["tickets"] = tickets
[docs] def get_tickets(self) -> list[object]: """ Retrieve the list of tickets. Returns: list[object]: The stored tickets list. Raises: KeyError: If no 'tickets' entry is found. """ return typing.cast(list, self.storage["tickets"])
[docs] def merge(self, storages: list[typing.Self]) -> None: """ Merge multiple KeyValueStore instances into this one. Combines lists for duplicate keys, overrides otherwise. Args: storages (list[KeyValueStore]): Other store instances to merge. """ for storage in storages: for key, value in storage.storage.items(): if ( key in self.storage and isinstance(self.storage[key], list) and isinstance(value, list) ): self.storage[key] += value else: self.storage[key] = value