Coverage for gws-app/gws/base/edit/helper.py: 64%

1from typing import Optional, cast 

2 

3import gws 

4import gws.base.action 

5import gws.base.feature 

6import gws.base.layer 

7import gws.base.legend 

8import gws.base.model 

9import gws.base.shape 

10import gws.base.template 

11import gws.base.web 

12import gws.lib.crs 

13import gws.gis.render 

14import gws.lib.image 

15import gws.lib.jsonx 

16import gws.lib.mime 

17 

18from . import api 

19 

20gws.ext.new.helper('edit') 

21 

22LIST_VIEWS = ['title', 'label'] 

23DEFAULT_TOLERANCE = 10, gws.Uom.px 

24 

25 

26class Object(gws.Node): 

27 

28 def get_models(self, req: gws.WebRequester, p: api.GetModelsRequest) -> list[gws.Model]: 

29 project = req.user.require_project(p.projectUid) 

30 return self.root.app.modelMgr.editable_models(project, req.user) 

31 

32 def get_models_response(self, req: gws.WebRequester, p: gws.Request, models: list[gws.Model]) -> api.GetModelsResponse: 

33 return api.GetModelsResponse( 

34 models=gws.u.compact(gws.props_of(m, req.user) for m in models) 

35 ) 

36 

37 ## 

38 

39 def get_features(self, req: gws.WebRequester, p: api.GetFeaturesRequest) -> list[gws.Feature]: 

40 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editList) 

41 

42 search = gws.SearchQuery(project=mc.project, tolerance=DEFAULT_TOLERANCE) 

43 if p.extent: 

44 search.bounds = gws.Bounds(crs=p.crs or mc.project.map.bounds.crs, extent=p.extent) 

45 if p.shapes: 

46 shapes = [gws.base.shape.from_props(s) for s in p.shapes] 

47 search.shape = shapes[0] if len(shapes) == 1 else shapes[0].union(shapes[1:]) 

48 if p.resolution: 

49 search.resolution = p.resolution 

50 if p.keyword: 

51 search.keyword = p.keyword 

52 if p.featureUids: 

53 search.uids = p.featureUids 

54 

55 fs = [] 

56 

57 for model_uid in p.modelUids: 

58 model = self.require_model(model_uid, req.user, gws.Access.read) 

59 fs.extend(model.find_features(search, mc)) 

60 

61 return fs 

62 

63 def get_features_response(self, req: gws.WebRequester, p: gws.Request, features: list[gws.Feature]) -> api.GetFeaturesResponse: 

64 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editList) 

65 return api.GetFeaturesResponse(features=self.feature_list_to_props(features, mc)) 

66 

67 ## 

68 

69 def get_relatable_features(self, req: gws.WebRequester, p: api.GetRelatableFeaturesRequest) -> list[gws.Feature]: 

70 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editList, max_depth=0) 

71 

72 model = self.require_model(p.modelUid, req.user, gws.Access.read) 

73 field = self.require_field(model, p.fieldName, req.user, gws.Access.read) 

74 search = gws.SearchQuery(keyword=p.keyword) 

75 

76 return field.find_relatable_features(search, mc) 

77 

78 def get_relatable_features_response(self, req: gws.WebRequester, p: gws.Request, features: list[gws.Feature]) -> api.GetRelatableFeaturesResponse: 

79 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editList) 

80 return api.GetRelatableFeaturesResponse(features=self.feature_list_to_props(features, mc)) 

81 

82 ## 

83 

84 def get_feature(self, req: gws.WebRequester, p: api.GetFeatureRequest) -> Optional[gws.Feature]: 

85 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editForm) 

86 model = self.require_model(p.modelUid, req.user, gws.Access.read) 

87 fs = model.get_features([p.featureUid], mc) 

88 if fs: 

89 return fs[0] 

90 

91 def get_feature_response(self, req: gws.WebRequester, p: gws.Request, feature: Optional[gws.Feature]) -> api.GetFeatureResponse: 

92 if not feature: 

93 raise gws.NotFoundError() 

94 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editForm) 

95 return api.GetFeatureResponse(feature=self.feature_to_props(feature, mc)) 

96 

97 ## 

98 

99 def init_feature(self, req: gws.WebRequester, p: api.InitFeatureRequest) -> gws.Feature: 

100 mc = self.model_context(req, p, gws.ModelOperation.create) 

101 

102 f = self.feature_from_props(p.feature, gws.Access.create, mc) 

103 f.createWithFeatures = [ 

104 self.feature_from_props(r, gws.Access.read, mc) 

105 for r in (p.feature.createWithFeatures or []) 

106 ] 

107 

108 f.model.init_feature(f, mc) 

109 return f 

110 

111 def init_feature_response(self, req: gws.WebRequester, p: gws.Request, feature: Optional[gws.Feature]) -> api.InitFeatureResponse: 

112 if not feature: 

113 raise gws.NotFoundError() 

114 mc = self.model_context(req, p, gws.ModelOperation.create) 

115 return api.InitFeatureResponse(feature=self.feature_to_props(feature, mc)) 

116 

117 ## 

118 

119 def write_feature(self, req: gws.WebRequester, p: api.WriteFeatureRequest) -> Optional[gws.Feature]: 

120 is_new = p.feature.isNew 

121 mc = self.model_context(req, p, gws.ModelOperation.create if is_new else gws.ModelOperation.update) 

122 

123 f = self.feature_from_props(p.feature, gws.Access.write, mc) 

124 f.createWithFeatures = [ 

125 self.feature_from_props(r, gws.Access.read, mc) 

126 for r in (p.feature.createWithFeatures or []) 

127 ] 

128 

129 if not f.model.validate_feature(f, mc): 

130 return f 

131 

132 if is_new: 

133 uid = f.model.create_feature(f, mc) 

134 else: 

135 uid = f.model.update_feature(f, mc) 

136 

137 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editForm) 

138 f_created = f.model.get_features([uid], mc) 

139 if not f_created: 

140 return 

141 

142 return f_created[0] 

143 

144 def write_feature_response(self, req: gws.WebRequester, p: api.WriteFeatureRequest, feature: Optional[gws.Feature]) -> api.WriteFeatureResponse: 

145 if not feature: 

146 raise gws.NotFoundError() 

147 if feature.errors: 

148 return api.WriteFeatureResponse(validationErrors=feature.errors) 

149 

150 mc = self.model_context(req, p, gws.ModelOperation.read, gws.ModelReadTarget.editForm) 

151 return api.WriteFeatureResponse( 

152 feature=self.feature_to_props(feature, mc), 

153 validationErrors=[] 

154 ) 

155 

156 ## 

157 

158 def delete_feature(self, req: gws.WebRequester, p: api.DeleteFeatureRequest) -> Optional[gws.Feature]: 

159 mc = self.model_context(req, p, gws.ModelOperation.delete) 

160 f = self.feature_from_props(p.feature, gws.Access.delete, mc) 

161 if f: 

162 f.model.delete_feature(f, mc) 

163 return f 

164 

165 def delete_feature_response(self, req: gws.WebRequester, p: api.DeleteFeatureRequest, feature: Optional[gws.Feature]) -> api.DeleteFeatureResponse: 

166 return api.DeleteFeatureResponse() 

167 

168 ## 

169 

170 def require_model(self, model_uid, user: gws.User, access: gws.Access) -> gws.Model: 

171 model = cast(gws.Model, user.acquire(model_uid, gws.ext.object.model, access)) 

172 if not model: 

173 raise gws.ForbiddenError(f'model {model_uid!r} not found or not accessible') 

174 if not model.isEditable: 

175 raise gws.ForbiddenError(f'model {model_uid!r} is not editable') 

176 return model 

177 

178 def require_field(self, model: gws.Model, field_name: str, user: gws.User, access: gws.Access) -> gws.ModelField: 

179 field = model.field(field_name) 

180 if not field: 

181 raise gws.ForbiddenError(f'field {field_name!r} not found in model {model.uid!r}') 

182 if not user.can(access, field): 

183 raise gws.ForbiddenError(f'field {field_name!r} not accessible for user {user.uid!r}') 

184 return field 

185 

186 def feature_from_props(self, props: gws.FeatureProps, access: gws.Access, mc: gws.ModelContext) -> gws.Feature: 

187 model = self.require_model(props.modelUid, mc.user, access) 

188 feature = model.feature_from_props(props, mc) 

189 if not feature: 

190 raise gws.NotFoundError() 

191 return feature 

192 

193 def feature_list_to_props(self, features: list[gws.Feature], mc: gws.ModelContext) -> list[gws.FeatureProps]: 

194 template_map = {} 

195 

196 for f in gws.base.model.iter_features(features, mc): 

197 if f.model.uid not in template_map: 

198 template_map[f.model.uid] = gws.u.compact( 

199 f.model.root.app.templateMgr.find_template( 

200 f'feature.{v}', 

201 [f.model, f.model.parent, mc.project], user=mc.user 

202 ) 

203 for v in LIST_VIEWS 

204 ) 

205 

206 f.render_views(template_map[f.model.uid], user=mc.user, project=mc.project) 

207 if mc.project.map: 

208 f.transform_to(mc.project.map.bounds.crs) 

209 

210 return [f.model.feature_to_props(f, mc) for f in features] 

211 

212 def feature_to_props(self, feature: gws.Feature, mc: gws.ModelContext) -> gws.FeatureProps: 

213 ps = self.feature_list_to_props([feature], mc) 

214 return ps[0] 

215 

216 def model_context(self, req: gws.WebRequester, p: gws.Request, op, target: Optional[gws.ModelReadTarget] = None, max_depth=1): 

217 return gws.ModelContext( 

218 op=op, 

219 target=target, 

220 user=req.user, 

221 project=req.user.require_project(p.projectUid), 

222 maxDepth=max_depth, 

223 )