Coverage for gws-app/gws/plugin/auth_session_manager/sqlite/__init__.py: 0%

1from typing import Optional 

2 

3import gws 

4import gws.base.auth 

5import gws.lib.datetimex 

6import gws.lib.jsonx 

7import gws.lib.osx 

8import gws.lib.sqlitex 

9 

10gws.ext.new.authSessionManager('sqlite') 

11 

12 

13class Config(gws.base.auth.session_manager.Config): 

14 """Configuration for sqlite sessions""" 

15 

16 path: Optional[str] 

17 """session storage path""" 

18 

19 

20_TABLE = 'sessions81' 

21 

22_INIT_DDL = f''' 

23 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

24 uid TEXT NOT NULL PRIMARY KEY, 

25 method_uid TEXT NOT NULL, 

26 user_uid TEXT NOT NULL, 

27 str_user TEXT NOT NULL, 

28 str_data TEXT NOT NULL, 

29 created INTEGER NOT NULL, 

30 updated INTEGER NOT NULL 

31 ) 

32''' 

33 

34 

35class Object(gws.base.auth.session_manager.Object): 

36 dbPath: str 

37 

38 def configure(self): 

39 self.dbPath = self.cfg('path', default=f'{gws.c.MISC_DIR}/sessions81.sqlite') 

40 

41 ## 

42 

43 def cleanup(self): 

44 last_time = gws.u.stime() - self.lifeTime 

45 self._db().execute(f'DELETE FROM {_TABLE} WHERE updated < :last_time', last_time=last_time) 

46 

47 def create(self, method, user, data=None): 

48 am = self.root.app.authMgr 

49 uid = gws.u.random_string(64) 

50 

51 self._db().insert(_TABLE, dict( 

52 uid=uid, 

53 method_uid=method.uid, 

54 user_uid=user.uid, 

55 str_user=am.serialize_user(user), 

56 str_data=gws.lib.jsonx.to_string(data or {}), 

57 created=gws.u.stime(), 

58 updated=gws.u.stime(), 

59 )) 

60 

61 return self.get(uid) 

62 

63 def delete(self, sess): 

64 self._db().execute(f'DELETE FROM {_TABLE} WHERE uid=:uid', uid=sess.uid) 

65 

66 def delete_all(self): 

67 self._db().execute(f'DELETE FROM {_TABLE}') 

68 

69 def get(self, uid): 

70 rs = self._db().select(f'SELECT * FROM {_TABLE} WHERE uid=:uid', uid=uid) 

71 if len(rs) == 1: 

72 return self._session(rs[0]) 

73 

74 def get_valid(self, uid): 

75 last_time = gws.u.stime() - self.lifeTime 

76 rs = self._db().select(f'SELECT * FROM {_TABLE} WHERE uid=:uid', uid=uid) 

77 if len(rs) == 1: 

78 rec = gws.u.to_dict(rs[0]) 

79 if rec['updated'] >= last_time: 

80 return self._session(rec) 

81 

82 def get_all(self): 

83 return [ 

84 self._session(rec) 

85 for rec in self._db().select(f'SELECT * FROM {_TABLE}') 

86 ] 

87 

88 def save(self, sess): 

89 if not sess.isChanged: 

90 return 

91 

92 self._db().execute( 

93 f'UPDATE {_TABLE} SET str_data=:str_data, updated=:updated WHERE uid=:uid', 

94 str_data=gws.lib.jsonx.to_string(sess.data or {}), 

95 updated=gws.u.stime(), 

96 uid=sess.uid 

97 ) 

98 

99 sess.isChanged = False 

100 

101 def touch(self, sess): 

102 if sess.isChanged: 

103 return self.save(sess) 

104 

105 self._db().execute( 

106 f'UPDATE {_TABLE} SET updated=:updated WHERE uid=:uid', 

107 updated=gws.u.stime(), 

108 uid=sess.uid 

109 ) 

110 

111 ## 

112 

113 def _session(self, rec): 

114 am = self.root.app.authMgr 

115 r = gws.u.to_dict(rec) 

116 return gws.base.auth.session.Object( 

117 uid=r['uid'], 

118 method=am.get_method(r['method_uid']), 

119 user=am.unserialize_user(r['str_user']), 

120 data=gws.lib.jsonx.from_string(r['str_data']), 

121 created=gws.lib.datetimex.from_timestamp(r['created']), 

122 updated=gws.lib.datetimex.from_timestamp(r['updated']), 

123 ) 

124 

125 _sqlitex: gws.lib.sqlitex.Object 

126 

127 def _db(self): 

128 if getattr(self, '_sqlitex', None) is None: 

129 self._sqlitex = gws.lib.sqlitex.Object(self.dbPath, _INIT_DDL) 

130 return self._sqlitex