Coverage for slidge/core/mixins/db.py: 96%

1import logging 

2import typing 

3from contextlib import contextmanager 

4 

5from ...db.models import Base, Contact, Participant, Room 

6 

7if typing.TYPE_CHECKING: 

8 from slidge import BaseGateway 

9 

10 

11class DBMixin: 

12 stored: Base 

13 xmpp: "BaseGateway" 

14 log: logging.Logger 

15 

16 def merge(self) -> None: 

17 with self.xmpp.store.session() as orm: 

18 self.stored = orm.merge(self.stored) 

19 

20 def commit(self, merge: bool = False) -> None: 

21 with self.xmpp.store.session(expire_on_commit=False) as orm: 

22 if merge: 

23 self.log.debug("Merging %s", self.stored) 

24 self.stored = orm.merge(self.stored) 

25 self.log.debug("Merged %s", self.stored) 

26 orm.add(self.stored) 

27 self.log.debug("Committing to DB") 

28 orm.commit() 

29 

30 

31class UpdateInfoMixin(DBMixin): 

32 """ 

33 This mixin just adds a context manager that prevents commiting to the DB 

34 on every attribute change. 

35 """ 

36 

37 stored: Contact | Room 

38 xmpp: "BaseGateway" 

39 log: logging.Logger 

40 

41 def __init__(self, *args, **kwargs) -> None: 

42 super().__init__(*args, **kwargs) 

43 self._updating_info = False 

44 if self.stored.extra_attributes is not None: 

45 self.deserialize_extra_attributes(self.stored.extra_attributes) 

46 

47 def serialize_extra_attributes(self) -> dict | None: 

48 return None 

49 

50 def deserialize_extra_attributes(self, data: dict) -> None: 

51 pass 

52 

53 @contextmanager 

54 def updating_info(self): 

55 self._updating_info = True 

56 yield 

57 self._updating_info = False 

58 self.stored.updated = True 

59 self.commit() 

60 

61 def commit(self, merge: bool = False) -> None: 

62 if self._updating_info: 

63 self.log.debug("Not updating %s right now", self.stored) 

64 else: 

65 self.stored.extra_attributes = self.serialize_extra_attributes() 

66 super().commit(merge=merge)