Coverage for gws-app/gws/base/action/manager.py: 53%

1"""Action manager and helpers for CLI commands.""" 

2 

3from typing import Optional 

4import gws 

5import gws.spec.runtime 

6 

7 

8def get_action_for_cli(root: gws.Root, action_name: str, project_uid: Optional[str] = None) -> Optional[gws.Action]: 

9 """Find an action for a CLI command, as the system user. 

10 

11 With a project uid, the action is searched in that project first. If it is not 

12 found there, or no project uid is given, it is searched among the application 

13 actions and then in all projects in turn. Errors are logged, not raised. 

14 

15 Args: 

16 root: The configuration root. 

17 action_name: Action type, for example ``alkis``. 

18 project_uid: Optional project uid. 

19 

20 Returns: 

21 The action object, or ``None`` if the project or the action is not found. 

22 """ 

23 

24 if project_uid: 

25 project = root.app.project(project_uid) 

26 if not project: 

27 gws.log.error(f'project {project_uid!r} not found') 

28 return None 

29 act = root.app.actionMgr.find_action(project, action_name, root.app.authMgr.systemUser) 

30 if act: 

31 return act 

32 gws.log.error(f'action {action_name!r} not found in {project_uid!r}') 

33 

34 act = root.app.actionMgr.find_action(None, action_name, root.app.authMgr.systemUser) 

35 if act: 

36 return act 

37 

38 for project in root.app.projects: 

39 act = root.app.actionMgr.find_action(project, action_name, root.app.authMgr.systemUser) 

40 if act: 

41 gws.log.info(f'using action {action_name!r} from {project.uid!r}') 

42 return act 

43 

44 gws.log.error(f'action {action_name!r} not found') 

45 

46 

47def parse_cli_request( 

48 root: gws.Root, 

49 command_category: gws.CommandCategory, 

50 command_name: str, 

51 params: dict, 

52 read_options=None, 

53): 

54 """Parse command parameters and return a command handler on a new action object. 

55 

56 The action class is instantiated directly, it is not a configured tree node. 

57 

58 Args: 

59 root: The configuration root. 

60 command_category: Command category. 

61 command_name: Command name. 

62 params: Raw command parameters. 

63 read_options: Options for the spec reader. 

64 

65 Returns: 

66 A tuple of the bound command method and the parsed request object. 

67 

68 Raises: 

69 gws.NotFoundError: If the command is not found. 

70 gws.BadRequestError: If the parameters cannot be parsed. 

71 """ 

72 

73 desc = root.specs.command_descriptor(command_category, command_name) 

74 if not desc: 

75 raise gws.NotFoundError(f'{command_category}.{command_name} not found') 

76 

77 try: 

78 request = root.specs.read(params, desc.tArg, options=read_options) 

79 except gws.spec.runtime.ReadError as exc: 

80 raise gws.BadRequestError(f'read error: {exc!r}') from exc 

81 

82 cls = gws.u.require(root.specs.get_class(desc.tOwner)) 

83 action = cls() 

84 

85 fn = getattr(action, desc.methodName) 

86 return fn, request 

87 

88 

89class Object(gws.ActionManager): 

90 """Action manager.""" 

91 

92 def actions_for_project(self, project, user): 

93 d = {} 

94 

95 for a in project.actions: 

96 if user.can_use(a): 

97 d[a.extType] = a 

98 

99 for a in self.root.app.actions: 

100 if a.extType not in d and user.can_use(a): 

101 d[a.extType] = a 

102 

103 return list(d.values()) 

104 

105 def prepare_action(self, command_category, command_name, params, path, user, read_options=None): 

106 desc = self.root.specs.command_descriptor(command_category, command_name) 

107 if not desc: 

108 raise gws.NotFoundError(f'{command_category}.{command_name} not found') 

109 

110 if desc.extCommandCategory == gws.CommandCategory.raw: 

111 project_uid = params.get('projectUid') 

112 

113 if not project_uid: 

114 path_parts = path.strip('/').split('/') 

115 try: 

116 project_uid = path_parts[path_parts.index('projectUid') + 1] 

117 except (ValueError, IndexError): 

118 pass 

119 

120 request = gws.Request( 

121 projectUid=project_uid, 

122 localeUid=params.get('localeUid'), 

123 ) 

124 else: 

125 try: 

126 request = gws.u.require(self.root.specs.read(params, desc.tArg, options=read_options)) 

127 except gws.spec.runtime.ReadError as exc: 

128 raise gws.BadRequestError(f'read error: {exc!r}') from exc 

129 

130 action = None 

131 

132 project_uid = request.get('projectUid') 

133 if project_uid: 

134 project = self.root.app.project(project_uid) 

135 if not project: 

136 raise gws.NotFoundError(f'project not found: {project_uid!r}') 

137 if not user.can_use(project): 

138 raise gws.ForbiddenError(f'project forbidden: {project_uid!r}') 

139 action = self._find_by_ext_name(project, desc.owner.extName, user) 

140 

141 if not action: 

142 action = self._find_by_ext_name(self.root.app, desc.owner.extName, user) 

143 

144 if not action: 

145 raise gws.NotFoundError(f'action {desc.owner.extName!r}: not found') 

146 

147 fn = getattr(action, desc.methodName) 

148 return fn, request 

149 

150 def find_action(self, project, ext_type, user): 

151 if project: 

152 a = self._find_by_ext_type(project, ext_type, user) 

153 if a: 

154 return a 

155 return self._find_by_ext_type(self.root.app, ext_type, user) 

156 

157 # @TODO build indexes for this 

158 

159 def _find_by_ext_name(self, obj, ext_name: str, user: gws.User): 

160 """Find an action of an object by its extension name, raise ``ForbiddenError`` if the user may not use it.""" 

161 for a in obj.actions: 

162 if a.extName == ext_name: 

163 if not user.can_use(a): 

164 raise gws.ForbiddenError(f'action {ext_name!r}: forbidden in {obj!r}') 

165 return a 

166 

167 def _find_by_ext_type(self, obj, ext_type: str, user: gws.User): 

168 """Find an action of an object by its extension type, raise ``ForbiddenError`` if the user may not use it.""" 

169 for a in obj.actions: 

170 if a.extType == ext_type: 

171 if not user.can_use(a): 

172 raise gws.ForbiddenError(f'action {ext_type!r}: forbidden in {obj!r}') 

173 return a