Coverage for gws-app/gws/base/storage/core.py: 52%

1"""Storage object.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.lib.jsonx 

7 

8 

9class Verb(gws.Enum): 

10 """Storage action verbs.""" 

11 

12 read = 'read' 

13 """Read an entry.""" 

14 write = 'write' 

15 """Write an entry.""" 

16 list = 'list' 

17 """List entries.""" 

18 delete = 'delete' 

19 """Delete an entry.""" 

20 

21 

22class State(gws.Data): 

23 """Storage state.""" 

24 

25 names: list[str] 

26 """List of entry names.""" 

27 canRead: bool 

28 """User can read entries.""" 

29 canWrite: bool 

30 """User can write entries.""" 

31 canCreate: bool 

32 """User can create entries.""" 

33 canDelete: bool 

34 """User can delete entries.""" 

35 

36 

37class Request(gws.Request): 

38 """Storage request.""" 

39 

40 verb: Verb 

41 """Action to perform.""" 

42 entryName: Optional[str] 

43 """Entry name, required for read, write and delete.""" 

44 entryData: Optional[dict] 

45 """Entry data to write.""" 

46 

47 

48class Response(gws.Response): 

49 """Storage response.""" 

50 

51 data: Optional[dict] 

52 """Entry data for a read request.""" 

53 state: State 

54 """Storage state after the request.""" 

55 

56 

57class Config(gws.ConfigWithAccess): 

58 """Storage for entries saved by users.""" 

59 

60 providerUid: Optional[str] 

61 """UID of the storage provider.""" 

62 categoryName: Optional[str] 

63 """Category under which entries are stored in the provider.""" 

64 

65 

66class Props(gws.Props): 

67 state: State 

68 

69 

70class Object(gws.Node): 

71 """Storage object, gives users access to entries of one category in a storage provider.""" 

72 

73 storageProvider: gws.StorageProvider 

74 """Storage provider.""" 

75 categoryName: str 

76 """Category under which entries are stored.""" 

77 

78 def configure(self): 

79 self.configure_provider() 

80 self.categoryName = self.cfg('categoryName') 

81 

82 def configure_provider(self): 

83 """Find the storage provider by ``providerUid``, or use the first provider. 

84 

85 Returns: 

86 ``True``. 

87 

88 Raises: 

89 ``gws.Error``: If no provider is found. 

90 """ 

91 self.storageProvider = self.root.app.storageMgr.find_provider(self.cfg('providerUid')) 

92 if not self.storageProvider: 

93 raise gws.Error(f'storage provider not found') 

94 return True 

95 

96 def props(self, user): 

97 return gws.Props( 

98 state=self.get_state_for(user), 

99 ) 

100 

101 def get_state_for(self, user): 

102 """Return the storage state for a user. 

103 

104 Args: 

105 user: The user. 

106 

107 Returns: 

108 The entry names, if the user can read them, and the user's permissions. 

109 """ 

110 return State( 

111 names=self.storageProvider.list_names(self.categoryName) if user.can_read(self) else [], 

112 canRead=user.can_read(self), 

113 canWrite=user.can_write(self), 

114 canDelete=user.can_delete(self), 

115 canCreate=user.can_create(self), 

116 ) 

117 

118 def handle_request(self, req: gws.WebRequester, p: Request) -> Response: 

119 """Handle a storage request from the client. 

120 

121 Writing an existing entry requires the ``write`` permission, writing a new 

122 one requires ``create``. Entry data is stored as JSON. 

123 

124 Args: 

125 req: Web requester. 

126 p: Storage request. 

127 

128 Returns: 

129 The entry data for a read request, and the new storage state. 

130 

131 Raises: 

132 ``gws.ForbiddenError``: If the user lacks the permission or the entry name or data is missing. 

133 """ 

134 state = self.get_state_for(req.user) 

135 data = None 

136 

137 if p.verb == Verb.list: 

138 if not state.canRead: 

139 raise gws.ForbiddenError() 

140 pass 

141 

142 if p.verb == Verb.read: 

143 if not state.canRead or not p.entryName: 

144 raise gws.ForbiddenError() 

145 rec = self.storageProvider.read(self.categoryName, p.entryName) 

146 if rec: 

147 data = gws.lib.jsonx.from_string(rec.data) 

148 

149 if p.verb == Verb.write: 

150 if not p.entryName or not p.entryData: 

151 raise gws.ForbiddenError() 

152 if p.entryName in state.names and not state.canWrite: 

153 raise gws.ForbiddenError() 

154 if p.entryName not in state.names and not state.canCreate: 

155 raise gws.ForbiddenError() 

156 

157 d = gws.lib.jsonx.to_string(p.entryData) 

158 self.storageProvider.write(self.categoryName, p.entryName, d, req.user.uid) 

159 

160 if p.verb == Verb.delete: 

161 if not state.canDelete or not p.entryName: 

162 raise gws.ForbiddenError() 

163 self.storageProvider.delete(self.categoryName, p.entryName) 

164 

165 return Response(data=data, state=self.get_state_for(req.user))