Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/session/storage.py: 76%

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3from abc import ABC, abstractmethod 

4from marshmallow import Schema, fields 

5 

6 

7class SessionStorageSchema(Schema): 

8 sessionid = fields.UUID() 

9 token = fields.Str() 

10 last_access = fields.Float() 

11 operator = fields.Bool() 

12 data = fields.Dict() 

13 

14 

15class SessionStorage(ABC): 

16 """The abstract session storage interface""" 

17 

18 _generate_token = None 

19 

20 @staticmethod 

21 def set_generate_token(fn): 

22 """Set a method from which to generate the session token 

23 

24 The function should return a token that can be stored into the session 

25 """ 

26 if not callable(fn): 

27 raise AttributeError("Token generation function must be callable") 

28 

29 SessionStorage._generate_token = fn 

30 

31 @abstractmethod 

32 def create(self, data): 

33 """Create a new session 

34 

35 Args: 

36 data (dict): Any data to store along side the basic session info 

37 

38 Returns: 

39 session: A `SessionStorageSchema` validated dict 

40 """ 

41 pass 

42 

43 @abstractmethod 

44 def remove(self, sessionid, token=None): 

45 """Remove a session 

46 

47 Args: 

48 sessionid (uuid): The sessionid to remove 

49 

50 Kwargs: 

51 token (str): The token to remove if sessionid is none 

52 

53 Returns: 

54 sessionid (uuid): The sessionid removed 

55 

56 """ 

57 pass 

58 

59 @abstractmethod 

60 def list(self): 

61 """List existing sessions 

62 

63 Returns: 

64 sessions (list(dict)): A list of `SessionStorageSchema` validated dicts 

65 """ 

66 pass 

67 

68 @abstractmethod 

69 def get(self, sessionid): 

70 """Get a session 

71 

72 Args: 

73 sessionid (uuid): The sessionid 

74 

75 Returns: 

76 session: A `SessionStorageSchema` validated dict 

77 """ 

78 pass 

79 

80 @abstractmethod 

81 def exists(self, sessionid): 

82 """Check if a session exists 

83 

84 Args: 

85 sessionid (uuid): The sessionid 

86 

87 Returns: 

88 True if the session exists, False if not 

89 """ 

90 pass 

91 

92 @abstractmethod 

93 def update(self, sessionid, **kwargs): 

94 """Update a a value in the data dict of a session 

95 

96 Updates a data property on the session and updates last_access time 

97 Args: 

98 sessionid (uuid): The sessionid 

99 

100 Kwargs: 

101 The properties in the data dict to update 

102 

103 """ 

104 pass 

105 

106 @abstractmethod 

107 def take_control(self, sessionid): 

108 """Set operator=True for the specified session 

109 

110 Sets the operator to True for this sessionid and to false for all the other existing 

111 sessions 

112 

113 Args: 

114 sessionid (uuid): The sessionid 

115 """ 

116 pass 

117 

118 @abstractmethod 

119 def yield_control(self, sessionid): 

120 """Set operator=False for the specified session 

121 

122 If the current session has control yield it. Return true if control was yielded 

123 

124 Args: 

125 sessionid (uuid): The sessionid 

126 """ 

127 pass 

128 

129 @abstractmethod 

130 def has_control(self, sessionid): 

131 """Check if this session has control 

132 

133 Args: 

134 sessionid (uuid): The sessionid 

135 

136 Returns: 

137 True if sessionid operator=True, False if not 

138 """ 

139 pass 

140 

141 @abstractmethod 

142 def no_control(self): 

143 """Check if someone in the application has control 

144 

145 Returns: 

146 True if no one has control 

147 """ 

148 pass