Skip to content

SQLAlchemyDB

hololinked.storage.SQLAlchemyDB

Bases: BaseSyncDB

SQLAlchemy-based database engine composed within Thing.

Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
 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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
class SQLAlchemyDB(BaseSyncDB):
    """SQLAlchemy-based database engine composed within `Thing`."""

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

        Highly unused and irrelevant currently.

        Returns
        -------
        `ThingInformation`

        Raises
        ------
        DatabaseError
        """
        if not inspect_database(self.engine).has_table("things"):
            return
        with self.sync_session() as session:
            stmt = select(ThingInformation).filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
            )
            data = session.execute(stmt)
            data = data.scalars().all()
            if len(data) == 0:
                return None
            elif len(data) == 1:
                return data[0]
            else:  # TODO raise different exception?
                raise DatabaseError(
                    "Multiple things with same instance name found, either cleanup database/detach/make new"
                )

    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
        ------
        DatabaseError
            if the property is not found in database or multiple properties with same name are found
        """
        with self.sync_session() as session:
            name = property if isinstance(property, str) else property.name
            stmt = select(SerializedProperty).filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
                name=name,
            )
            data = session.execute(stmt)
            prop = data.scalars().all()  # type: list[SerializedProperty]
            if len(prop) == 0:
                raise DatabaseError(f"property {name} not found in database")
            elif len(prop) > 1:
                raise DatabaseError("multiple properties with same name found")  # Impossible actually
            if not deserialized:
                return prop[0]
            serializer = Serializers.content_types.get(prop[0].content_type, None) or Serializers.for_object(
                self.thing.id,
                self.thing.__class__.__name__,
                name,
            )
            return serializer.loads(prop[0].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

        Raises
        ------
        DatabaseError
            if multiple properties with same name are found
        """
        name = property if isinstance(property, str) else property.name
        if self.in_batch_call_context:
            self._batch_call_context[threading.get_ident()][name] = value
            return
        with self.sync_session() as session:
            stmt = select(SerializedProperty).filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
                name=name,
            )
            data = session.execute(stmt)
            prop = data.scalars().all()
            if len(prop) > 1:
                raise DatabaseError("multiple properties with same name found")  # Impossible actually
            if len(prop) == 1:
                prop = prop[0]
                serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    name,
                )
                prop.serialized_value = serializer.dumps(value)
                prop.updated_at = datetime.now().isoformat()
                prop.content_type = serializer.content_type
            else:
                serializer = Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    name,
                )
                now = datetime.now().isoformat()
                prop = SerializedProperty(
                    id=None,  # ty: ignore[invalid-argument-type]
                    name=name,
                    serialized_value=serializer.dumps(value),
                    thing_id=self.thing.id,
                    thing_class=self.thing.__class__.__name__,
                    created_at=now,
                    updated_at=now,
                    content_type=serializer.content_type,
                )
                session.add(prop)
            session.commit()

    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
            deserilize the properties if True

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        with self.sync_session() as session:
            names = []
            for obj in properties:
                names.append(obj if isinstance(obj, str) else obj.name)
            stmt = (
                select(SerializedProperty)
                .filter_by(
                    thing_id=self.thing.id,
                    thing_class=self.thing.__class__.__name__,
                )
                .filter(SerializedProperty.name.in_(names))
            )
            data = session.execute(stmt)
            unserialized_props = data.scalars().all()
            props = dict()
            for prop in unserialized_props:
                serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    prop.name,
                )
                props[prop.name] = (
                    prop.serialized_value if not deserialized else serializer.loads(prop.serialized_value)
                )
            return props

    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

        Raises
        ------
        DatabaseError
            if multiple properties with same name are found
        """
        if self.in_batch_call_context:
            for obj, value in properties.items():
                name = obj if isinstance(obj, str) else obj.name
                self._batch_call_context[threading.get_ident()][name] = value
            return
        with self.sync_session() as session:
            names = []
            for obj in properties:
                names.append(obj if isinstance(obj, str) else obj.name)
            stmt = (
                select(SerializedProperty)
                .filter_by(
                    thing_id=self.thing.id,
                    thing_class=self.thing.__class__.__name__,
                )
                .filter(SerializedProperty.name.in_(names))
            )
            data = session.execute(stmt)
            db_props = data.scalars().all()
            for obj, value in properties.items():
                name = obj if isinstance(obj, str) else obj.name
                db_prop = list(filter(lambda db_prop: db_prop.name == name, db_props))  # type: list[SerializedProperty]
                if len(db_prop) > 1:
                    raise DatabaseError("multiple properties with same name found")  # Impossible actually
                if len(db_prop) == 1:
                    db_prop = db_prop[0]  # type: SerializedProperty
                    serializer = Serializers.content_types.get(db_prop.content_type, None) or Serializers.for_object(
                        self.thing.id,
                        self.thing.__class__.__name__,
                        name,
                    )
                    db_prop.serialized_value = serializer.dumps(value)
                    db_prop.updated_at = datetime.now().isoformat()
                    db_prop.content_type = serializer.content_type
                else:
                    serializer = Serializers.for_object(
                        self.thing.id,
                        self.thing.__class__.__name__,
                        name,
                    )
                    now = datetime.now().isoformat()
                    prop = SerializedProperty(
                        id=None,  # ty: ignore[invalid-argument-type]
                        name=name,
                        serialized_value=serializer.dumps(value),
                        thing_id=self.thing.id,
                        thing_class=self.thing.__class__.__name__,
                        created_at=now,
                        updated_at=now,
                        content_type=serializer.content_type,
                    )
                    session.add(prop)
            session.commit()

    def get_all_properties(self, deserialized: bool = True) -> dict[str, Any] | Sequence[SerializedProperty]:
        """
        Get all properties stored in the database.

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

        Returns
        -------
        dict[str, Any]
            property names and values as items
        """
        with self.sync_session() as session:
            stmt = select(SerializedProperty).filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
            )
            data = session.execute(stmt)
            existing_props = data.scalars().all()  # type: list[SerializedProperty]
            if not deserialized:
                return existing_props
            props = dict()
            for prop in existing_props:
                serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    prop.name,
                )
                props[prop.name] = serializer.loads(prop.serialized_value)
            return props

    def create_missing_properties(
        self,
        properties: dict[str, Property],
        get_missing_property_names: bool = False,
    ) -> None | list[str]:
        """
        Create any and all missing properties in the 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 = []
        with self.sync_session() as session:
            existing_props = self.get_all_properties()
            for prop in properties.values():
                if prop.name not in existing_props:
                    serializer = Serializers.for_object(
                        self.thing.id,
                        self.thing.__class__.__name__,
                        prop.name,
                    )
                    now = datetime.now().isoformat()
                    prop = SerializedProperty(
                        id=None,  # ty: ignore[invalid-argument-type]
                        name=prop.name,
                        serialized_value=serializer.dumps(getattr(self.thing, prop.name)),
                        thing_id=self.thing.id,
                        thing_class=self.thing.__class__.__name__,
                        created_at=now,
                        updated_at=now,
                        content_type=serializer.content_type,
                    )
                    session.add(prop)
                    missing_props.append(prop.name)
            session.commit()
        if get_missing_property_names:
            return missing_props

    def create_init_properties(self, thing_id: str, thing_class: str, **properties: Any) -> None:
        """
        Create properties that are supposed to be initialized from database for a thing instance.

        Invoke this method once before running the thing instance to store its initial value.

        Parameters
        ----------
        thing_id: str
            ID of the thing instance to which these properties belong
        thing_class: str
            Class name of the thing instance to which these properties belong
        properties: dict[str, Any]
            property names and their initial values as dictionary pairs
        """
        with self.sync_session() as session:
            for name, value in properties.items():
                serializer = Serializers.for_object(thing_id, thing_class, name)
                now = datetime.now().isoformat()
                prop = SerializedProperty(
                    id=None,  # ty: ignore[invalid-argument-type]
                    name=name,
                    serialized_value=serializer.dumps(value),
                    thing_id=thing_id,
                    thing_class=thing_class,
                    created_at=now,
                    updated_at=now,
                    content_type=serializer.content_type,
                )
                session.add(prop)
            session.commit()

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')))

engine instance-attribute 

engine = create_engine(URL)

sync_session instance-attribute 

sync_session = sessionmaker(engine, expire_on_commit=True)

in_batch_call_context property 

in_batch_call_context

Functions

__init__ 

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

Initialize BaseSyncDB for a Thing instance.

Parameters:

Name Type Description Default
thing
Thing

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

 required
config_file
str | None

Path to the database configuration file. sqlite backend with default settings will be used if not provided.

 None

Raises:

Type Description
ValueError

If the loaded configuration is not a valid config for SQLAlchemy-based database engines (but a valid config of another type).
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(self, thing: Thing, config_file: str | None = None) -> None:
    """
    Initialize BaseSyncDB for a Thing instance.

    Parameters
    ----------
    thing: Thing
        The `Thing` instance which uses this database engine for configuration storage.
    config_file: str, optional
        Path to the database configuration file. `sqlite` backend with default settings will be used if not provided.

    Raises
    ------
    ValueError
        If the loaded configuration is not a valid config for SQLAlchemy-based database engines
        (but a valid config of another type).
    """
    super().__init__(thing=thing, config_file=config_file)
    if not isinstance(self.config, (SQLDBConfig, SQLiteConfig)):
        raise ValueError(f"You might have provided invalid SQLAlchemy config. Loaded config: {self.config}")
    self.engine = create_engine(self.config.URL)
    self.sync_session = sessionmaker(self.engine, expire_on_commit=True)
    if self.config.provider == "sqlite":
        ThingTableBase.metadata.create_all(self.engine)

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
`ThingInformation`

Raises:

Type Description
DatabaseError
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
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
def fetch_own_info(self):  # -> ThingInformation:
    """
    Fetch `Thing` instance's own information (some useful metadata which helps the `Thing` run).

    Highly unused and irrelevant currently.

    Returns
    -------
    `ThingInformation`

    Raises
    ------
    DatabaseError
    """
    if not inspect_database(self.engine).has_table("things"):
        return
    with self.sync_session() as session:
        stmt = select(ThingInformation).filter_by(
            thing_id=self.thing.id,
            thing_class=self.thing.__class__.__name__,
        )
        data = session.execute(stmt)
        data = data.scalars().all()
        if len(data) == 0:
            return None
        elif len(data) == 1:
            return data[0]
        else:  # TODO raise different exception?
            raise DatabaseError(
                "Multiple things with same instance name found, either cleanup database/detach/make new"
            )

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
DatabaseError

if the property is not found in database or multiple properties with same name are found
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
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
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
    ------
    DatabaseError
        if the property is not found in database or multiple properties with same name are found
    """
    with self.sync_session() as session:
        name = property if isinstance(property, str) else property.name
        stmt = select(SerializedProperty).filter_by(
            thing_id=self.thing.id,
            thing_class=self.thing.__class__.__name__,
            name=name,
        )
        data = session.execute(stmt)
        prop = data.scalars().all()  # type: list[SerializedProperty]
        if len(prop) == 0:
            raise DatabaseError(f"property {name} not found in database")
        elif len(prop) > 1:
            raise DatabaseError("multiple properties with same name found")  # Impossible actually
        if not deserialized:
            return prop[0]
        serializer = Serializers.content_types.get(prop[0].content_type, None) or Serializers.for_object(
            self.thing.id,
            self.thing.__class__.__name__,
            name,
        )
        return serializer.loads(prop[0].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

deserilize the properties if True

 True

Returns:

Type Description
dict[str, Any]

property names and values as items
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
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
        deserilize the properties if True

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    with self.sync_session() as session:
        names = []
        for obj in properties:
            names.append(obj if isinstance(obj, str) else obj.name)
        stmt = (
            select(SerializedProperty)
            .filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
            )
            .filter(SerializedProperty.name.in_(names))
        )
        data = session.execute(stmt)
        unserialized_props = data.scalars().all()
        props = dict()
        for prop in unserialized_props:
            serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                self.thing.id,
                self.thing.__class__.__name__,
                prop.name,
            )
            props[prop.name] = (
                prop.serialized_value if not deserialized else serializer.loads(prop.serialized_value)
            )
        return props

get_all_properties 

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

Get all properties stored in the database.

Parameters:

Name Type Description Default
deserialized
bool

deserilize the properties if True

 True

Returns:

Type Description
dict[str, Any]

property names and values as items
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def get_all_properties(self, deserialized: bool = True) -> dict[str, Any] | Sequence[SerializedProperty]:
    """
    Get all properties stored in the database.

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

    Returns
    -------
    dict[str, Any]
        property names and values as items
    """
    with self.sync_session() as session:
        stmt = select(SerializedProperty).filter_by(
            thing_id=self.thing.id,
            thing_class=self.thing.__class__.__name__,
        )
        data = session.execute(stmt)
        existing_props = data.scalars().all()  # type: list[SerializedProperty]
        if not deserialized:
            return existing_props
        props = dict()
        for prop in existing_props:
            serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                self.thing.id,
                self.thing.__class__.__name__,
                prop.name,
            )
            props[prop.name] = serializer.loads(prop.serialized_value)
        return props

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

Raises:

Type Description
DatabaseError

if multiple properties with same name are found
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
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
227
228
229
230
231
232
233
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

    Raises
    ------
    DatabaseError
        if multiple properties with same name are found
    """
    name = property if isinstance(property, str) else property.name
    if self.in_batch_call_context:
        self._batch_call_context[threading.get_ident()][name] = value
        return
    with self.sync_session() as session:
        stmt = select(SerializedProperty).filter_by(
            thing_id=self.thing.id,
            thing_class=self.thing.__class__.__name__,
            name=name,
        )
        data = session.execute(stmt)
        prop = data.scalars().all()
        if len(prop) > 1:
            raise DatabaseError("multiple properties with same name found")  # Impossible actually
        if len(prop) == 1:
            prop = prop[0]
            serializer = Serializers.content_types.get(prop.content_type, None) or Serializers.for_object(
                self.thing.id,
                self.thing.__class__.__name__,
                name,
            )
            prop.serialized_value = serializer.dumps(value)
            prop.updated_at = datetime.now().isoformat()
            prop.content_type = serializer.content_type
        else:
            serializer = Serializers.for_object(
                self.thing.id,
                self.thing.__class__.__name__,
                name,
            )
            now = datetime.now().isoformat()
            prop = SerializedProperty(
                id=None,  # ty: ignore[invalid-argument-type]
                name=name,
                serialized_value=serializer.dumps(value),
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
                created_at=now,
                updated_at=now,
                content_type=serializer.content_type,
            )
            session.add(prop)
        session.commit()

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

Raises:

Type Description
DatabaseError

if multiple properties with same name are found
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
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

    Raises
    ------
    DatabaseError
        if multiple properties with same name are found
    """
    if self.in_batch_call_context:
        for obj, value in properties.items():
            name = obj if isinstance(obj, str) else obj.name
            self._batch_call_context[threading.get_ident()][name] = value
        return
    with self.sync_session() as session:
        names = []
        for obj in properties:
            names.append(obj if isinstance(obj, str) else obj.name)
        stmt = (
            select(SerializedProperty)
            .filter_by(
                thing_id=self.thing.id,
                thing_class=self.thing.__class__.__name__,
            )
            .filter(SerializedProperty.name.in_(names))
        )
        data = session.execute(stmt)
        db_props = data.scalars().all()
        for obj, value in properties.items():
            name = obj if isinstance(obj, str) else obj.name
            db_prop = list(filter(lambda db_prop: db_prop.name == name, db_props))  # type: list[SerializedProperty]
            if len(db_prop) > 1:
                raise DatabaseError("multiple properties with same name found")  # Impossible actually
            if len(db_prop) == 1:
                db_prop = db_prop[0]  # type: SerializedProperty
                serializer = Serializers.content_types.get(db_prop.content_type, None) or Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    name,
                )
                db_prop.serialized_value = serializer.dumps(value)
                db_prop.updated_at = datetime.now().isoformat()
                db_prop.content_type = serializer.content_type
            else:
                serializer = Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    name,
                )
                now = datetime.now().isoformat()
                prop = SerializedProperty(
                    id=None,  # ty: ignore[invalid-argument-type]
                    name=name,
                    serialized_value=serializer.dumps(value),
                    thing_id=self.thing.id,
                    thing_class=self.thing.__class__.__name__,
                    created_at=now,
                    updated_at=now,
                    content_type=serializer.content_type,
                )
                session.add(prop)
        session.commit()

create_missing_properties 

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

Create any and all missing properties in the 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/sqlalchemydb.py 
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def create_missing_properties(
    self,
    properties: dict[str, Property],
    get_missing_property_names: bool = False,
) -> None | list[str]:
    """
    Create any and all missing properties in the 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 = []
    with self.sync_session() as session:
        existing_props = self.get_all_properties()
        for prop in properties.values():
            if prop.name not in existing_props:
                serializer = Serializers.for_object(
                    self.thing.id,
                    self.thing.__class__.__name__,
                    prop.name,
                )
                now = datetime.now().isoformat()
                prop = SerializedProperty(
                    id=None,  # ty: ignore[invalid-argument-type]
                    name=prop.name,
                    serialized_value=serializer.dumps(getattr(self.thing, prop.name)),
                    thing_id=self.thing.id,
                    thing_class=self.thing.__class__.__name__,
                    created_at=now,
                    updated_at=now,
                    content_type=serializer.content_type,
                )
                session.add(prop)
                missing_props.append(prop.name)
        session.commit()
    if get_missing_property_names:
        return missing_props

create_init_properties 

create_init_properties(thing_id: str, thing_class: str, **properties: Any) -> None

Create properties that are supposed to be initialized from database for a thing instance.

Invoke this method once before running the thing instance to store its initial value.

Parameters:

Name Type Description Default
thing_id
str

ID of the thing instance to which these properties belong

 required
thing_class
str

Class name of the thing instance to which these properties belong

 required
properties
Any

property names and their initial values as dictionary pairs

 {}
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def create_init_properties(self, thing_id: str, thing_class: str, **properties: Any) -> None:
    """
    Create properties that are supposed to be initialized from database for a thing instance.

    Invoke this method once before running the thing instance to store its initial value.

    Parameters
    ----------
    thing_id: str
        ID of the thing instance to which these properties belong
    thing_class: str
        Class name of the thing instance to which these properties belong
    properties: dict[str, Any]
        property names and their initial values as dictionary pairs
    """
    with self.sync_session() as session:
        for name, value in properties.items():
            serializer = Serializers.for_object(thing_id, thing_class, name)
            now = datetime.now().isoformat()
            prop = SerializedProperty(
                id=None,  # ty: ignore[invalid-argument-type]
                name=name,
                serialized_value=serializer.dumps(value),
                thing_id=thing_id,
                thing_class=thing_class,
                created_at=now,
                updated_at=now,
                content_type=serializer.content_type,
            )
            session.add(prop)
        session.commit()

batch_db_commit

Write multiple properties to a database at once.

Useful for optimizing sequential sets/writes of multiple properties which are stored onto a database.

Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
class batch_db_commit:
    """
    Write multiple properties to a database at once.

    Useful for optimizing sequential sets/writes of multiple properties which are stored onto a database.
    """

    def __init__(self, db_engine: SQLAlchemyDB) -> None:
        self.db_engine = db_engine

    def __enter__(self) -> None:
        self.db_engine._batch_call_context[threading.get_ident()] = dict()

    def __exit__(self, exc_type, exc_value, exc_tb) -> None:
        data = self.db_engine._batch_call_context.pop(threading.get_ident(), dict())
        if exc_type is None:
            self.db_engine.set_properties(data)
            return
        for name, value in data.items():
            try:
                self.db_engine.set_property(name, value)
            except Exception as ex:
                self.db_engine.thing.logger.error(
                    f"failed to set property {name} to value {value} during batch commit due to exception {ex}"
                )

Functions

__init__ 

__init__(db_engine: SQLAlchemyDB) -> None
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
465
466
def __init__(self, db_engine: SQLAlchemyDB) -> None:
    self.db_engine = db_engine

BaseSyncDB

Bases: BaseDB

Base class for a synchronous (blocking) database engine, implements sqlalchemy engine & session creation.

Default DB engine for Thing & called immediately after properties are set/written.

Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
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
class BaseSyncDB(BaseDB):
    """
    Base class for a synchronous (blocking) database engine, implements sqlalchemy engine & session creation.

    Default DB engine for `Thing` & called immediately after properties are set/written.
    """

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

        Parameters
        ----------
        thing: Thing
            The `Thing` instance which uses this database engine for configuration storage.
        config_file: str, optional
            Path to the database configuration file. `sqlite` backend with default settings will be used if not provided.

        Raises
        ------
        ValueError
            If the loaded configuration is not a valid config for SQLAlchemy-based database engines
            (but a valid config of another type).
        """
        super().__init__(thing=thing, config_file=config_file)
        if not isinstance(self.config, (SQLDBConfig, SQLiteConfig)):
            raise ValueError(f"You might have provided invalid SQLAlchemy config. Loaded config: {self.config}")
        self.engine = create_engine(self.config.URL)
        self.sync_session = sessionmaker(self.engine, expire_on_commit=True)
        if self.config.provider == "sqlite":
            ThingTableBase.metadata.create_all(self.engine)

Attributes

engine instance-attribute 

engine = create_engine(URL)

sync_session instance-attribute 

sync_session = sessionmaker(engine, expire_on_commit=True)

Functions

__init__ 

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

Initialize BaseSyncDB for a Thing instance.

Parameters:

Name Type Description Default
thing
Thing

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

 required
config_file
str | None

Path to the database configuration file. sqlite backend with default settings will be used if not provided.

 None

Raises:

Type Description
ValueError

If the loaded configuration is not a valid config for SQLAlchemy-based database engines (but a valid config of another type).
Source code in repo/hololinked/hololinked/storage/sqlalchemydb.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(self, thing: Thing, config_file: str | None = None) -> None:
    """
    Initialize BaseSyncDB for a Thing instance.

    Parameters
    ----------
    thing: Thing
        The `Thing` instance which uses this database engine for configuration storage.
    config_file: str, optional
        Path to the database configuration file. `sqlite` backend with default settings will be used if not provided.

    Raises
    ------
    ValueError
        If the loaded configuration is not a valid config for SQLAlchemy-based database engines
        (but a valid config of another type).
    """
    super().__init__(thing=thing, config_file=config_file)
    if not isinstance(self.config, (SQLDBConfig, SQLiteConfig)):
        raise ValueError(f"You might have provided invalid SQLAlchemy config. Loaded config: {self.config}")
    self.engine = create_engine(self.config.URL)
    self.sync_session = sessionmaker(self.engine, expire_on_commit=True)
    if self.config.provider == "sqlite":
        ThingTableBase.metadata.create_all(self.engine)