Skip to content

MongoDB

hololinked.storage.MongoDB

Bases: BaseDB

Provides persistence for Thing properties using MongoDB.

Properties are stored in a 'properties' collection, with fields:

  • id: Thing instance identifier
  • name: property name
  • serialized_value: serialized property value
Source code in repo/hololinked/hololinked/storage/mongodb.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class MongoDB(BaseDB):
    """
    Provides persistence for Thing properties using MongoDB.

    Properties are stored in a 'properties' collection, with fields:

    - id: Thing instance identifier
    - name: property name
    - serialized_value: serialized property value
    """

    def __init__(self, thing: Thing, config_file: str) -> None:
        """
        Initialize MongoDB for a Thing instance.

        Connects to MongoDB and sets up collections.

        Parameters
        ----------
        thing: Thing
            The `Thing` instance which uses this database engine for configuration storage.
        config_file: str
            Path to the MongoDB configuration file.

        Raises
        ------
        ValueError
            If the loaded configuration is not a valid Mongo DB Config (but a valid config of another type).
        """
        super().__init__(thing=thing, config_file=config_file)
        if not isinstance(self.config, MongoDBConfig):
            raise ValueError(f"You might have provided invalid MongoDB config. Loaded config: {self.config}")
        self.client = MongoClient(self.config.URL)
        self.db = self.client[self.config.database]
        self.properties = self.db["properties"]
        self.things = self.db["things"]

    def fetch_own_info(self):
        """
        Fetch `Thing` instance's own information (some useful metadata which helps the `Thing` run).

        Highly unused and irrelevant currently.

        Returns
        -------
        dict[str, Any] | None
            Metadata document for the Thing instance, or None if not found.
        """
        doc = self.things.find_one({"id": self.thing.id})
        return doc

    def get_property(self, property: str | Property, deserialized: bool = True) -> Any:
        """
        Fetch a single property value from the database.

        Parameters
        ----------
        property: str | Property
            string name or descriptor object
        deserialized: bool, default True
            deserialize the property if True

        Returns
        -------
        Any
            property value

        Raises
        ------
        PyMongoError
            if the property is not found in database
        """
        name = property if isinstance(property, str) else property.name
        doc = self.properties.find_one({"id": self.thing.id, "name": name})
        if not doc:
            raise mongo_errors.PyMongoError(f"property {name} not found in database")
        if not deserialized:
            return doc
        serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
        return serializer.loads(base64.b64decode(doc["serialized_value"]))

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

        Parameters
        ----------
        property: str | Property
            string name or descriptor object
        value: Any
            value of the property
        """
        name = property if isinstance(property, str) else property.name
        serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
        serialized_value = base64.b64encode(serializer.dumps(value)).decode("utf-8")
        self.properties.update_one(
            {"id": self.thing.id, "name": name}, {"$set": {"serialized_value": serialized_value}}, upsert=True
        )

    def get_properties(self, properties: dict[str | Property, Any], deserialized: bool = True) -> dict[str, Any]:
        """
        Get multiple properties at once from the database.

        Parameters
        ----------
        properties: List[str | Property]
            string names or the descriptor of the properties as a list
        deserialized: bool, default True
            deserialize the properties if True

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        names = [obj if isinstance(obj, str) else obj.name for obj in properties.keys()]
        cursor = self.properties.find({"id": self.thing.id, "name": {"$in": names}})
        result = {}
        for doc in cursor:
            serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, doc["name"])
            result[doc["name"]] = (
                doc["serialized_value"]
                if not deserialized
                else serializer.loads(base64.b64decode(doc["serialized_value"]))
            )
        return result

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

        Parameters
        ----------
        properties: Dict[str | Property, Any]
            string names or the descriptor of the property and any value as dictionary pairs
        """
        for obj, value in properties.items():
            name = obj if isinstance(obj, str) else obj.name
            serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
            serialized_value = base64.b64encode(serializer.dumps(value)).decode("utf-8")
            self.properties.update_one(
                {"id": self.thing.id, "name": name},
                {"$set": {"serialized_value": serialized_value}},
                upsert=True,
            )

    def get_all_properties(self, deserialized: bool = True) -> dict[str, Any]:
        """
        Get all properties of the `Thing` instance stored in the database.

        Parameters
        ----------
        deserialized: bool, default True
            deserialize the properties if True

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        cursor = self.properties.find({"id": self.thing.id})
        result = {}
        for doc in cursor:
            serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, doc["name"])
            result[doc["name"]] = (
                doc["serialized_value"]
                if not deserialized
                else serializer.loads(base64.b64decode(doc["serialized_value"]))
            )
        return result

    def create_missing_properties(
        self,
        properties: dict[str, Property],
        get_missing_property_names: bool = False,
    ) -> Any:
        """
        Create any and all missing properties of `Thing` instance in database.

        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 = []
        existing_props = self.get_all_properties()
        for name, new_prop in properties.items():
            if name not in existing_props:
                serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, new_prop.name)
                serialized_value = base64.b64encode(serializer.dumps(getattr(self.thing, new_prop.name))).decode(
                    "utf-8"
                )
                self.properties.insert_one(
                    {"id": self.thing.id, "name": new_prop.name, "serialized_value": serialized_value}
                )
                missing_props.append(name)
        if get_missing_property_names:
            return missing_props

Attributes

thing instance-attribute 

thing = thing

config instance-attribute 

config = load_conf(config_file=config_file, default_file_path=join(TEMP_DIR_DB, get_sanitized_filename_from_random_string(f'{__name__}.{id}', extension='db')))

client instance-attribute 

client = MongoClient(URL)

db instance-attribute 

db = client[database]

properties instance-attribute 

properties = db['properties']

things instance-attribute 

things = db['things']

in_batch_call_context property 

in_batch_call_context

Functions

__init__ 

__init__(thing: Thing, config_file: str) -> None

Initialize MongoDB for a Thing instance.

Connects to MongoDB and sets up collections.

Parameters:

Name Type Description Default
thing
Thing

The Thing instance which uses this database engine for configuration storage.

 required
config_file
str

Path to the MongoDB configuration file.

 required

Raises:

Type Description
ValueError

If the loaded configuration is not a valid Mongo DB Config (but a valid config of another type).
Source code in repo/hololinked/hololinked/storage/mongodb.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, thing: Thing, config_file: str) -> None:
    """
    Initialize MongoDB for a Thing instance.

    Connects to MongoDB and sets up collections.

    Parameters
    ----------
    thing: Thing
        The `Thing` instance which uses this database engine for configuration storage.
    config_file: str
        Path to the MongoDB configuration file.

    Raises
    ------
    ValueError
        If the loaded configuration is not a valid Mongo DB Config (but a valid config of another type).
    """
    super().__init__(thing=thing, config_file=config_file)
    if not isinstance(self.config, MongoDBConfig):
        raise ValueError(f"You might have provided invalid MongoDB config. Loaded config: {self.config}")
    self.client = MongoClient(self.config.URL)
    self.db = self.client[self.config.database]
    self.properties = self.db["properties"]
    self.things = self.db["things"]

from_thing classmethod 

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

Build the database engine for a Thing, taking its configuration from the db_config_file argument.

Parameters:

Name Type Description Default
thing
Thing

the Thing instance whose configuration is stored in this database

 required
kwargs
Any

the keyword arguments given to the Thing; db_config_file is read from here and the rest ignored

 {}

Returns:

Type Description
Self

the database engine, connected and ready to use
Source code in repo/hololinked/hololinked/storage/bases.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_thing(cls, thing: Thing, **kwargs: Any) -> Self:
    """
    Build the database engine for a `Thing`, taking its configuration from the `db_config_file` argument.

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

    Returns
    -------
    Self
        the database engine, connected and ready to use
    """
    return cls(thing=thing, config_file=kwargs.get("db_config_file"))

fetch_own_info 

fetch_own_info()

Fetch Thing instance's own information (some useful metadata which helps the Thing run).

Highly unused and irrelevant currently.

Returns:

Type Description
dict[str, Any] | None

Metadata document for the Thing instance, or None if not found.
Source code in repo/hololinked/hololinked/storage/mongodb.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
def fetch_own_info(self):
    """
    Fetch `Thing` instance's own information (some useful metadata which helps the `Thing` run).

    Highly unused and irrelevant currently.

    Returns
    -------
    dict[str, Any] | None
        Metadata document for the Thing instance, or None if not found.
    """
    doc = self.things.find_one({"id": self.thing.id})
    return doc

get_property 

get_property(property: str | Property, deserialized: bool = True) -> Any

Fetch a single property value from the database.

Parameters:

Name Type Description Default
property
str | Property

string name or descriptor object

 required
deserialized
bool

deserialize the property if True

 True

Returns:

Type Description
Any

property value

Raises:

Type Description
PyMongoError

if the property is not found in database
Source code in repo/hololinked/hololinked/storage/mongodb.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_property(self, property: str | Property, deserialized: bool = True) -> Any:
    """
    Fetch a single property value from the database.

    Parameters
    ----------
    property: str | Property
        string name or descriptor object
    deserialized: bool, default True
        deserialize the property if True

    Returns
    -------
    Any
        property value

    Raises
    ------
    PyMongoError
        if the property is not found in database
    """
    name = property if isinstance(property, str) else property.name
    doc = self.properties.find_one({"id": self.thing.id, "name": name})
    if not doc:
        raise mongo_errors.PyMongoError(f"property {name} not found in database")
    if not deserialized:
        return doc
    serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
    return serializer.loads(base64.b64decode(doc["serialized_value"]))

get_properties 

get_properties(properties: dict[str | Property, Any], deserialized: bool = True) -> dict[str, Any]

Get multiple properties at once from the database.

Parameters:

Name Type Description Default
properties
dict[str | Property, Any]

string names or the descriptor of the properties as a list

 required
deserialized
bool

deserialize the properties if True

 True

Returns:

Type Description
dict[str, Any]

property names and values as items
Source code in repo/hololinked/hololinked/storage/mongodb.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_properties(self, properties: dict[str | Property, Any], deserialized: bool = True) -> dict[str, Any]:
    """
    Get multiple properties at once from the database.

    Parameters
    ----------
    properties: List[str | Property]
        string names or the descriptor of the properties as a list
    deserialized: bool, default True
        deserialize the properties if True

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    names = [obj if isinstance(obj, str) else obj.name for obj in properties.keys()]
    cursor = self.properties.find({"id": self.thing.id, "name": {"$in": names}})
    result = {}
    for doc in cursor:
        serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, doc["name"])
        result[doc["name"]] = (
            doc["serialized_value"]
            if not deserialized
            else serializer.loads(base64.b64decode(doc["serialized_value"]))
        )
    return result

get_all_properties 

get_all_properties(deserialized: bool = True) -> dict[str, Any]

Get all properties of the Thing instance stored in the database.

Parameters:

Name Type Description Default
deserialized
bool

deserialize the properties if True

 True

Returns:

Type Description
dict[str, Any]

property names and values as items
Source code in repo/hololinked/hololinked/storage/mongodb.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def get_all_properties(self, deserialized: bool = True) -> dict[str, Any]:
    """
    Get all properties of the `Thing` instance stored in the database.

    Parameters
    ----------
    deserialized: bool, default True
        deserialize the properties if True

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    cursor = self.properties.find({"id": self.thing.id})
    result = {}
    for doc in cursor:
        serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, doc["name"])
        result[doc["name"]] = (
            doc["serialized_value"]
            if not deserialized
            else serializer.loads(base64.b64decode(doc["serialized_value"]))
        )
    return result

set_property 

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

Change the value of an already existing property in the database.

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/mongodb.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def set_property(self, property: str | Property, value: Any) -> None:
    """
    Change the value of an already existing property in the database.

    Parameters
    ----------
    property: str | Property
        string name or descriptor object
    value: Any
        value of the property
    """
    name = property if isinstance(property, str) else property.name
    serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
    serialized_value = base64.b64encode(serializer.dumps(value)).decode("utf-8")
    self.properties.update_one(
        {"id": self.thing.id, "name": name}, {"$set": {"serialized_value": serialized_value}}, upsert=True
    )

set_properties 

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

Change the values of already existing properties at once in the database.

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/mongodb.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def set_properties(self, properties: dict[str | Property, Any]) -> None:
    """
    Change the values of already existing properties at once in the database.

    Parameters
    ----------
    properties: Dict[str | Property, Any]
        string names or the descriptor of the property and any value as dictionary pairs
    """
    for obj, value in properties.items():
        name = obj if isinstance(obj, str) else obj.name
        serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, name)
        serialized_value = base64.b64encode(serializer.dumps(value)).decode("utf-8")
        self.properties.update_one(
            {"id": self.thing.id, "name": name},
            {"$set": {"serialized_value": serialized_value}},
            upsert=True,
        )

create_missing_properties 

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

Create any and all missing properties of Thing instance in database.

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/mongodb.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def create_missing_properties(
    self,
    properties: dict[str, Property],
    get_missing_property_names: bool = False,
) -> Any:
    """
    Create any and all missing properties of `Thing` instance in database.

    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 = []
    existing_props = self.get_all_properties()
    for name, new_prop in properties.items():
        if name not in existing_props:
            serializer = Serializers.for_object(self.thing.id, self.thing.__class__.__name__, new_prop.name)
            serialized_value = base64.b64encode(serializer.dumps(getattr(self.thing, new_prop.name))).decode(
                "utf-8"
            )
            self.properties.insert_one(
                {"id": self.thing.id, "name": new_prop.name, "serialized_value": serialized_value}
            )
            missing_props.append(name)
    if get_missing_property_names:
        return missing_props