JSONFileStorage

hololinked.storage.JSONFileStorage

Bases: BaseConfigurationRepository

JSON file based storage engine composed within Thing.

Carries out property operations such as storing and retrieving values from a plain JSON file.

Source code in repo/hololinked/hololinked/storage/jsonfile.py

class JSONFileStorage(BaseConfigurationRepository):
    """
    JSON file based storage engine composed within `Thing`.

    Carries out property operations such as storing and retrieving values from a plain JSON file.
    """

    def __init__(self, thing: Thing, filename: str, serializer: BaseSerializer | None = None):
        """
        Initialize JSONFileStorage for a Thing instance.

        Parameters
        ----------
        thing: Thing
            The `Thing` instance which uses this storage for configuration storage.
        filename: str
            Path to the JSON file to use for storage.
        serializer: BaseSerializer | None, optional
            Serializer for encoding and decoding JSON data. Defaults to an instance of `JSONSerializer`.
        """
        self.filename = filename
        self.thing = thing
        self._serializer = serializer or JSONSerializer()
        self._lock = threading.RLock()
        self._data = self._load()

    @classmethod
    def from_thing(cls, thing: Thing, **kwargs: Any) -> Self:
        """
        Build the JSON file storage for a `Thing`, deriving the file name from its ID if none was given.

        The file always lives under `global_config.TEMP_DIR_DB`, so a relative `json_filename` cannot escape it.

        Parameters
        ----------
        thing: Thing
            the `Thing` instance whose configuration is stored in the file
        kwargs: dict[str, Any]
            the keyword arguments given to the `Thing`; `json_filename` is read from here and the rest ignored

        Returns
        -------
        Self
            the storage engine, with the file loaded if it already exists
        """
        filename = kwargs.get("json_filename") or get_sanitized_filename_from_random_string(
            thing.id,
            extension="json",
        )
        return cls(thing=thing, filename=os.path.join(global_config.TEMP_DIR_DB, filename))

    def _load(self) -> dict[str, Any]:
        """
        Load and decode data from the JSON file.

        Returns
        -------
        dict[str, Any]
            A dictionary of all stored properties. Empty if the file does not exist or cannot be decoded.
        """
        if not os.path.exists(self.filename) or os.path.getsize(self.filename) == 0:
            return {}
        with open(self.filename, "rb") as f:
            raw_bytes = f.read()
            if not raw_bytes:
                return {}
            return self._serializer.loads(raw_bytes)  # type: ignore[invalid-return-type]

    def _save(self):
        """Encode and write data to the JSON file."""
        raw_bytes = self._serializer.dumps(self._data)
        with open(self.filename, "wb") as f:
            f.write(raw_bytes)

    def get_property(self, property: str | Property, **kwargs) -> Any:  # ty: ignore[invalid-method-override]
        """
        Fetch a single property value from the JSON file.

        Parameters
        ----------
        property: str | Property
            string name or descriptor object
        deserialized: bool, default True
            Ignored for this storage engine since JSON file storage is always deserialized.
            Included for interface compatibility.

        Returns
        -------
        Any
            property value

        Raises
        ------
        KeyError
            If the property is not found in storage.
        """
        name = property if isinstance(property, str) else property.name
        if name not in self._data:
            raise KeyError(f"property {name} not found in JSON storage")
        with self._lock:
            return self._data[name]

    def set_property(self, property: str | Property, value: Any) -> None:
        """
        Change the value of an already existing property in the JSON file.

        Parameters
        ----------
        property: str | Property
            string name or descriptor object
        value: Any
            value of the property
        """
        name = property if isinstance(property, str) else property.name
        with self._lock:
            self._data[name] = value
            self._save()

    def get_properties(
        self,
        properties: dict[str | Property, Any],
        **kwargs,
    ) -> dict[str, Any]:  # ty: ignore[invalid-method-override]
        """
        Get multiple properties at once from the JSON file.

        Parameters
        ----------
        properties: List[str | Property]
            string names or the descriptor of the properties as a list
        deserialized: bool, default True
            Ignored for this storage engine since JSON file storage is always deserialized.
            Included for interface compatibility.

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        names = [key if isinstance(key, str) else key.name for key in properties.keys()]
        with self._lock:
            return {name: self._data.get(name) for name in names}

    def set_properties(self, properties: dict[str | Property, Any]) -> None:
        """
        Change the values of already existing properties at once in the JSON file.

        Parameters
        ----------
        properties: Dict[str | Property, Any]
            string names or the descriptor of the property and any value as dictionary pairs
        """
        with self._lock:
            for obj, value in properties.items():
                name = obj if isinstance(obj, str) else obj.name
                self._data[name] = value
            self._save()

    def get_all_properties(self, **kwargs) -> dict[str, Any]:  # ty: ignore[invalid-method-override]
        """
        Get all properties stored in the JSON file.

        Parameters
        ----------
        deserialized: bool, default True
            Ignored for this storage engine since JSON file storage is always deserialized.
            Included for interface compatibility.

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        with self._lock:
            return dict(self._data)

    def create_missing_properties(
        self,
        properties: dict[str, Property],
        get_missing_property_names: bool = False,
    ) -> list[str] | None:
        """
        Create any and all missing properties in the JSON file.

        Parameters
        ----------
        properties: Dict[str, Property]
            descriptors of the properties
        get_missing_property_names: bool, default False
            whether to return the list of missing property names

        Returns
        -------
        List[str]
            list of missing properties if get_missing_property_names is True
        """
        missing_props = []
        with self._lock:
            existing_props = self.get_all_properties()
            for name, new_prop in properties.items():
                if name not in existing_props:
                    self._data[name] = getattr(self.thing, new_prop.name)
                    missing_props.append(name)
            self._save()
        if get_missing_property_names:
            return missing_props

Attributes

thing instance-attribute

thing = thing

filename instance-attribute

filename = filename

Functions

__init__

__init__(thing: Thing, filename: str, serializer: BaseSerializer | None = None)

Initialize JSONFileStorage for a Thing instance.

Parameters:

Name	Type	Description	Default
thing	Thing	The Thing instance which uses this storage for configuration storage.	required
filename	str	Path to the JSON file to use for storage.	required
serializer	BaseSerializer | None	Serializer for encoding and decoding JSON data. Defaults to an instance of JSONSerializer.	None

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def __init__(self, thing: Thing, filename: str, serializer: BaseSerializer | None = None):
    """
    Initialize JSONFileStorage for a Thing instance.

    Parameters
    ----------
    thing: Thing
        The `Thing` instance which uses this storage for configuration storage.
    filename: str
        Path to the JSON file to use for storage.
    serializer: BaseSerializer | None, optional
        Serializer for encoding and decoding JSON data. Defaults to an instance of `JSONSerializer`.
    """
    self.filename = filename
    self.thing = thing
    self._serializer = serializer or JSONSerializer()
    self._lock = threading.RLock()
    self._data = self._load()

from_thing classmethod

from_thing(thing: Thing, **kwargs: Any) -> Self

Build the JSON file storage for a Thing, deriving the file name from its ID if none was given.

The file always lives under global_config.TEMP_DIR_DB, so a relative json_filename cannot escape it.

Parameters:

Name	Type	Description	Default
thing	Thing	the Thing instance whose configuration is stored in the file	required
kwargs	Any	the keyword arguments given to the Thing; json_filename is read from here and the rest ignored	{}

Returns:

Type	Description
Self	the storage engine, with the file loaded if it already exists

Source code in repo/hololinked/hololinked/storage/jsonfile.py

@classmethod
def from_thing(cls, thing: Thing, **kwargs: Any) -> Self:
    """
    Build the JSON file storage for a `Thing`, deriving the file name from its ID if none was given.

    The file always lives under `global_config.TEMP_DIR_DB`, so a relative `json_filename` cannot escape it.

    Parameters
    ----------
    thing: Thing
        the `Thing` instance whose configuration is stored in the file
    kwargs: dict[str, Any]
        the keyword arguments given to the `Thing`; `json_filename` is read from here and the rest ignored

    Returns
    -------
    Self
        the storage engine, with the file loaded if it already exists
    """
    filename = kwargs.get("json_filename") or get_sanitized_filename_from_random_string(
        thing.id,
        extension="json",
    )
    return cls(thing=thing, filename=os.path.join(global_config.TEMP_DIR_DB, filename))

get_property

get_property(property: str | Property, **kwargs) -> Any

Fetch a single property value from the JSON file.

Parameters:

Name	Type	Description	Default
property	str | Property	string name or descriptor object	required
deserialized		Ignored for this storage engine since JSON file storage is always deserialized. Included for interface compatibility.	required

Returns:

Type	Description
Any	property value

Raises:

Type	Description
KeyError	If the property is not found in storage.

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def get_property(self, property: str | Property, **kwargs) -> Any:  # ty: ignore[invalid-method-override]
    """
    Fetch a single property value from the JSON file.

    Parameters
    ----------
    property: str | Property
        string name or descriptor object
    deserialized: bool, default True
        Ignored for this storage engine since JSON file storage is always deserialized.
        Included for interface compatibility.

    Returns
    -------
    Any
        property value

    Raises
    ------
    KeyError
        If the property is not found in storage.
    """
    name = property if isinstance(property, str) else property.name
    if name not in self._data:
        raise KeyError(f"property {name} not found in JSON storage")
    with self._lock:
        return self._data[name]

get_properties

get_properties(properties: dict[str | Property, Any], **kwargs) -> dict[str, Any]

Get multiple properties at once from the JSON file.

Parameters:

Name	Type	Description	Default
properties	dict[str | Property, Any]	string names or the descriptor of the properties as a list	required
deserialized		Ignored for this storage engine since JSON file storage is always deserialized. Included for interface compatibility.	required

Returns:

Type	Description
dict[str, Any]	property names and values as items

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def get_properties(
    self,
    properties: dict[str | Property, Any],
    **kwargs,
) -> dict[str, Any]:  # ty: ignore[invalid-method-override]
    """
    Get multiple properties at once from the JSON file.

    Parameters
    ----------
    properties: List[str | Property]
        string names or the descriptor of the properties as a list
    deserialized: bool, default True
        Ignored for this storage engine since JSON file storage is always deserialized.
        Included for interface compatibility.

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    names = [key if isinstance(key, str) else key.name for key in properties.keys()]
    with self._lock:
        return {name: self._data.get(name) for name in names}

get_all_properties

get_all_properties(**kwargs) -> dict[str, Any]

Get all properties stored in the JSON file.

Parameters:

Name	Type	Description	Default
deserialized		Ignored for this storage engine since JSON file storage is always deserialized. Included for interface compatibility.	required

Returns:

Type	Description
dict[str, Any]	property names and values as items

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def get_all_properties(self, **kwargs) -> dict[str, Any]:  # ty: ignore[invalid-method-override]
    """
    Get all properties stored in the JSON file.

    Parameters
    ----------
    deserialized: bool, default True
        Ignored for this storage engine since JSON file storage is always deserialized.
        Included for interface compatibility.

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    with self._lock:
        return dict(self._data)

set_property

set_property(property: str | Property, value: Any) -> None

Change the value of an already existing property in the JSON file.

Parameters:

Name	Type	Description	Default
property	str | Property	string name or descriptor object	required
value	Any	value of the property	required

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def set_property(self, property: str | Property, value: Any) -> None:
    """
    Change the value of an already existing property in the JSON file.

    Parameters
    ----------
    property: str | Property
        string name or descriptor object
    value: Any
        value of the property
    """
    name = property if isinstance(property, str) else property.name
    with self._lock:
        self._data[name] = value
        self._save()

set_properties

set_properties(properties: dict[str | Property, Any]) -> None

Change the values of already existing properties at once in the JSON file.

Parameters:

Name	Type	Description	Default
properties	dict[str | Property, Any]	string names or the descriptor of the property and any value as dictionary pairs	required

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def set_properties(self, properties: dict[str | Property, Any]) -> None:
    """
    Change the values of already existing properties at once in the JSON file.

    Parameters
    ----------
    properties: Dict[str | Property, Any]
        string names or the descriptor of the property and any value as dictionary pairs
    """
    with self._lock:
        for obj, value in properties.items():
            name = obj if isinstance(obj, str) else obj.name
            self._data[name] = value
        self._save()

create_missing_properties

create_missing_properties(properties: dict[str, Property], get_missing_property_names: bool = False) -> list[str] | None

Create any and all missing properties in the JSON file.

Parameters:

Name	Type	Description	Default
properties	dict[str, Property]	descriptors of the properties	required
get_missing_property_names	bool	whether to return the list of missing property names	False

Returns:

Type	Description
List[str]	list of missing properties if get_missing_property_names is True

Source code in repo/hololinked/hololinked/storage/jsonfile.py

def create_missing_properties(
    self,
    properties: dict[str, Property],
    get_missing_property_names: bool = False,
) -> list[str] | None:
    """
    Create any and all missing properties in the JSON file.

    Parameters
    ----------
    properties: Dict[str, Property]
        descriptors of the properties
    get_missing_property_names: bool, default False
        whether to return the list of missing property names

    Returns
    -------
    List[str]
        list of missing properties if get_missing_property_names is True
    """
    missing_props = []
    with self._lock:
        existing_props = self.get_all_properties()
        for name, new_prop in properties.items():
            if name not in existing_props:
                self._data[name] = getattr(self.thing, new_prop.name)
                missing_props.append(name)
        self._save()
    if get_missing_property_names:
        return missing_props