Coverage for gws-app/gws/base/exporter/worker.py: 81%

1"""Export job worker.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.job 

7 

8 

9class Object(gws.base.job.worker.Object): 

10 """Export job worker. 

11 

12 Loads the requested features, runs the exporter and stores the export 

13 result in the job. Progress is reported in percent, one step for loading, 

14 one per feature and one for writing. 

15 """ 

16 

17 project: gws.Project 

18 """Project of the request.""" 

19 request: gws.ExportRequest 

20 """Export request.""" 

21 result: gws.ExportResult 

22 """Export result, filled by the exporter.""" 

23 progressTotal: int = 1 

24 """Total number of progress steps.""" 

25 progressStep: int = 0 

26 """Current progress step.""" 

27 progressPercent: int = 0 

28 """Last reported progress, in percent.""" 

29 

30 @classmethod 

31 def run(cls, root: gws.Root, job: gws.Job): 

32 """Run an export job. 

33 

34 Called by the job manager. Reads the request from the file given in the job payload. 

35 

36 Args: 

37 root: Root object. 

38 job: The job. 

39 """ 

40 request = gws.u.unserialize_from_path(job.payload.get('requestPath')) 

41 w = cls(root, job.user, job, request) 

42 w.work() 

43 

44 def __init__(self, root: gws.Root, user: gws.User, job: Optional[gws.Job], request: gws.ExportRequest): 

45 """Create a worker. 

46 

47 Args: 

48 root: Root object. 

49 user: User the export runs for. 

50 job: The job, or ``None`` to run without a job. 

51 request: Export request. 

52 """ 

53 super().__init__(root, user, job) 

54 self.request = request 

55 self.result = gws.ExportResult( 

56 path='', 

57 mimeType='', 

58 numFiles=0, 

59 numFeaturesTotal=0, 

60 numFeaturesExported=0, 

61 errors=[], 

62 ) 

63 

64 def work(self): 

65 """Run the export and store the result in the job, if there is one. 

66 

67 Raises: 

68 ``gws.NotFoundError``: If the project or the exporter does not exist. 

69 ``gws.ForbiddenError``: If the user cannot read the project. 

70 ``gws.JobTerminated``: If the job was cancelled or removed. 

71 """ 

72 self.project = cast(gws.Project, self.user.require(self.request.projectUid, gws.ext.object.project)) 

73 

74 exporter = self.root.app.exporterMgr.get_exporter( 

75 [self.project, self.root.app], 

76 self.request.exporterUid, 

77 self.user, 

78 ) 

79 if not exporter: 

80 raise gws.NotFoundError('exporter not found') 

81 

82 # load + features + write 

83 self.progressTotal = 1 + len(self.request.features or []) + 1 

84 self.update_job(numSteps=100) 

85 

86 features = self.load_features() 

87 self.notify('load') 

88 

89 ea = gws.ExportArgs( 

90 exporter=exporter, 

91 features=features, 

92 shape=self.request.shape, 

93 project=self.project, 

94 user=self.user, 

95 notify=self.notify, 

96 ) 

97 

98 exporter.run(ea, self.result) 

99 self.update_job(state=gws.JobState.complete, result=self.result) 

100 

101 def load_features(self): 

102 """Load the requested features. 

103 

104 Features of models with the ``client`` export strategy are built from 

105 the request props, others are read from their models by uid. Features 

106 of unknown models and features that are not found are skipped with a 

107 warning. 

108 

109 Returns: 

110 The features, in the order of the request. 

111 """ 

112 if not self.request.features: 

113 return [] 

114 

115 self.result.numFeaturesTotal = len(self.request.features) 

116 

117 models: dict[str, gws.Model] = {} 

118 uid_map: dict[str, set[str]] = {} 

119 fs_map: dict[str, gws.Feature] = {} 

120 

121 mc = gws.ModelContext(op=gws.ModelOperation.export, user=self.user, project=self.project) 

122 

123 for fp in self.request.features: 

124 model_uid = fp.modelUid 

125 feature_uid = fp.uid 

126 

127 model = models.get(model_uid) 

128 if not model: 

129 model = self.root.app.modelMgr.get_model(model_uid, self.user) 

130 if not model: 

131 gws.log.warning(f'model not found: {model_uid}') 

132 continue 

133 models[model_uid] = model 

134 

135 guid = gws.u.join_uid(model_uid, feature_uid) 

136 es = model.exportStrategy or gws.FeatureExportStrategy.load 

137 if es == gws.FeatureExportStrategy.client: 

138 fs_map[guid] = model.feature_from_props(fp, mc) 

139 else: 

140 uid_map.setdefault(model_uid, set()).add(feature_uid) 

141 

142 for model_uid, feature_uids in uid_map.items(): 

143 model = models[model_uid] 

144 fs = model.get_features(feature_uids, mc) 

145 for f in fs: 

146 guid = gws.u.join_uid(model_uid, f.uid()) 

147 fs_map[guid] = f 

148 

149 fs = [] 

150 

151 for fp in self.request.features: 

152 guid = gws.u.join_uid(fp.modelUid, fp.uid) 

153 f = fs_map.get(guid) 

154 if not f: 

155 gws.log.warning(f'feature not found: {guid}') 

156 continue 

157 fs.append(f) 

158 

159 return fs 

160 

161 def notify(self, event, details=None): 

162 """Advance the progress by one step and update the job. 

163 

164 Args: 

165 event: Event name, not used. 

166 details: Event details, not used. 

167 

168 Raises: 

169 ``gws.JobTerminated``: If the job was cancelled or removed. 

170 """ 

171 job = self.get_job() 

172 if not job: 

173 return 

174 self.progressStep += 1 

175 p = int(self.progressStep / self.progressTotal * 100) 

176 if p > self.progressPercent: 

177 self.progressPercent = p 

178 self.update_job(numSteps=100, step=self.progressPercent)