Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/contrib/sessions/backends/db.py: 29%

1import logging 

2 

3from django.contrib.sessions.backends.base import CreateError, SessionBase, UpdateError 

4from django.core.exceptions import SuspiciousOperation 

5from django.db import DatabaseError, IntegrityError, router, transaction 

6from django.utils import timezone 

7from django.utils.functional import cached_property 

8 

9 

10class SessionStore(SessionBase): 

11 """ 

12 Implement database session store. 

13 """ 

14 

15 def __init__(self, session_key=None): 

16 super().__init__(session_key) 

17 

18 @classmethod 

19 def get_model_class(cls): 

20 # Avoids a circular import and allows importing SessionStore when 

21 # django.contrib.sessions is not in INSTALLED_APPS. 

22 from django.contrib.sessions.models import Session 

23 

24 return Session 

25 

26 @cached_property 

27 def model(self): 

28 return self.get_model_class() 

29 

30 def _get_session_from_db(self): 

31 try: 

32 return self.model.objects.get( 

33 session_key=self.session_key, expire_date__gt=timezone.now() 

34 ) 

35 except (self.model.DoesNotExist, SuspiciousOperation) as e: 

36 if isinstance(e, SuspiciousOperation): 

37 logger = logging.getLogger("django.security.%s" % e.__class__.__name__) 

38 logger.warning(str(e)) 

39 self._session_key = None 

40 

41 def load(self): 

42 s = self._get_session_from_db() 

43 return self.decode(s.session_data) if s else {} 

44 

45 def exists(self, session_key): 

46 return self.model.objects.filter(session_key=session_key).exists() 

47 

48 def create(self): 

49 while True: 

50 self._session_key = self._get_new_session_key() 

51 try: 

52 # Save immediately to ensure we have a unique entry in the 

53 # database. 

54 self.save(must_create=True) 

55 except CreateError: 

56 # Key wasn't unique. Try again. 

57 continue 

58 self.modified = True 

59 return 

60 

61 def create_model_instance(self, data): 

62 """ 

63 Return a new instance of the session model object, which represents the 

64 current session state. Intended to be used for saving the session data 

65 to the database. 

66 """ 

67 return self.model( 

68 session_key=self._get_or_create_session_key(), 

69 session_data=self.encode(data), 

70 expire_date=self.get_expiry_date(), 

71 ) 

72 

73 def save(self, must_create=False): 

74 """ 

75 Save the current session data to the database. If 'must_create' is 

76 True, raise a database error if the saving operation doesn't create a 

77 new entry (as opposed to possibly updating an existing entry). 

78 """ 

79 if self.session_key is None: 

80 return self.create() 

81 data = self._get_session(no_load=must_create) 

82 obj = self.create_model_instance(data) 

83 using = router.db_for_write(self.model, instance=obj) 

84 try: 

85 with transaction.atomic(using=using): 

86 obj.save( 

87 force_insert=must_create, force_update=not must_create, using=using 

88 ) 

89 except IntegrityError: 

90 if must_create: 

91 raise CreateError 

92 raise 

93 except DatabaseError: 

94 if not must_create: 

95 raise UpdateError 

96 raise 

97 

98 def delete(self, session_key=None): 

99 if session_key is None: 

100 if self.session_key is None: 

101 return 

102 session_key = self.session_key 

103 try: 

104 self.model.objects.get(session_key=session_key).delete() 

105 except self.model.DoesNotExist: 

106 pass 

107 

108 @classmethod 

109 def clear_expired(cls): 

110 cls.get_model_class().objects.filter(expire_date__lt=timezone.now()).delete()