Coverage for gws-app/gws/base/job/manager.py: 69%

1"""Job manager.""" 

2 

3from typing import Optional 

4import sys 

5 

6import gws 

7import gws.lib.dynimport 

8import gws.lib.jsonx 

9import gws.lib.sqlitex 

10import gws.lib.datetimex as dtx 

11import gws.server.spool 

12 

13 

14class Object(gws.JobManager): 

15 """Job manager.""" 

16 

17 TABLE = 'jobs' 

18 """Name of the jobs table.""" 

19 DDL = f""" 

20 CREATE TABLE IF NOT EXISTS {TABLE} ( 

21 uid TEXT NOT NULL PRIMARY KEY, 

22 userUid TEXT DEFAULT '', 

23 userStr TEXT DEFAULT '', 

24 worker TEXT DEFAULT '', 

25 state TEXT DEFAULT '', 

26 error TEXT DEFAULT '', 

27 numSteps INTEGER DEFAULT 0, 

28 step INTEGER DEFAULT 0, 

29 stepName TEXT DEFAULT '', 

30 payload TEXT DEFAULT '', 

31 result TEXT DEFAULT '', 

32 created INTEGER DEFAULT 0, 

33 updated INTEGER DEFAULT 0 

34 ) 

35 """ 

36 """DDL statement for the jobs table.""" 

37 

38 dbPath: str 

39 """Path of the SQLite database.""" 

40 

41 def configure(self): 

42 ver = self.root.specs.version.rpartition('.')[0] 

43 self.dbPath = self.cfg('path', default=f'{gws.c.MISC_DIR}/jobs.{ver}.sqlite') 

44 

45 def create_job(self, worker, user, payload=None): 

46 job_uid = gws.u.random_string(64) 

47 gws.log.debug(f'JOB {job_uid}: creating: {worker=} {user.uid=}') 

48 

49 self._db().insert(self.TABLE, dict(uid=job_uid)) 

50 

51 mod_path = gws.u.require(sys.modules.get(worker.__module__)).__file__ 

52 self._write( 

53 job_uid, 

54 dict( 

55 userUid=user.uid, 

56 userStr=self.root.app.authMgr.serialize_user(user), 

57 worker=f'{mod_path}:{worker.__name__}', 

58 state=gws.JobState.open, 

59 payload=payload, 

60 result={}, 

61 created=gws.u.stime(), 

62 updated=gws.u.stime(), 

63 ), 

64 ) 

65 

66 return self._get_job_or_fail(job_uid) 

67 

68 def get_job(self, job_uid: str, user=None, state=None): 

69 job, msg = self._get_job(job_uid, user, state) 

70 if not job: 

71 gws.log.error(msg) 

72 return job 

73 

74 def _get_job_or_fail(self, job_uid: str, user=None, state=None): 

75 """Return a job like ``get_job``, but raise ``gws.Error`` if it is not found.""" 

76 job, msg = self._get_job(job_uid, user, state) 

77 if not job: 

78 raise gws.Error(msg) 

79 return job 

80 

81 def _get_job(self, job_uid: str, user=None, state=None): 

82 """Read a job, checking the user and state; return the job and an error message.""" 

83 rs = self._db().select(f'SELECT * FROM {self.TABLE} WHERE uid=:uid', uid=job_uid) 

84 if not rs: 

85 return None, f'JOB {job_uid}: not found' 

86 

87 rec = rs[0] 

88 

89 job_user = None 

90 us = rec.get('userStr') 

91 if us: 

92 job_user = self.root.app.authMgr.unserialize_user(us) 

93 if not job_user: 

94 job_user = self.root.app.authMgr.guestUser 

95 

96 if user and job_user.uid != user.uid: 

97 return None, f'JOB {job_uid}: wrong user {job_user.uid=} {user.uid=}' 

98 

99 if state and rec.get('state') != state: 

100 return None, f'JOB {job_uid}: wrong state {rec.get("state")=} {state=}' 

101 

102 job = gws.Job( 

103 uid=rec['uid'], 

104 user=job_user, 

105 worker=rec['worker'], 

106 state=rec['state'], 

107 error=rec['error'], 

108 numSteps=rec['numSteps'] or 0, 

109 step=rec['step'] or 0, 

110 stepName=rec['stepName'] or '', 

111 payload=gws.lib.jsonx.from_string(rec['payload'] or '{}'), 

112 result=gws.lib.jsonx.from_string(rec['result'] or '{}'), 

113 timeCreated=dtx.from_timestamp(rec['created'] or 0), 

114 timeUpdated=dtx.from_timestamp(rec['updated'] or 0), 

115 ) 

116 return job, '' 

117 

118 def update_job(self, job, **kwargs): 

119 job = self.get_job(job.uid) 

120 if not job: 

121 return 

122 self._write(job.uid, kwargs) 

123 return self._get_job_or_fail(job.uid) 

124 

125 def _write(self, job_uid, rec): 

126 """Update a job record, serializing the payload and the result.""" 

127 rec['updated'] = gws.u.stime() 

128 if 'payload' in rec: 

129 rec['payload'] = gws.lib.jsonx.to_string(rec['payload'] or {}) 

130 if 'result' in rec: 

131 rec['result'] = gws.lib.jsonx.to_string(rec['result'] or {}) 

132 

133 gws.log.debug(f'JOB {job_uid}: save {rec=}') 

134 self._db().update(self.TABLE, rec, job_uid) 

135 

136 def schedule_job(self, job: gws.Job): 

137 job = self._get_job_or_fail(job.uid, state=gws.JobState.open) 

138 if gws.server.spool.is_active(): 

139 gws.server.spool.add(job) 

140 return job 

141 return self.run_job(job) 

142 

143 def run_job(self, job: gws.Job): 

144 job_uid = job.uid 

145 

146 # atomically mark an 'open' job as 'running' 

147 

148 tmp = gws.u.random_string(64) 

149 self._db().execute( 

150 f'UPDATE {self.TABLE} SET state=:tmp WHERE uid=:uid AND state=:state', 

151 uid=job_uid, 

152 tmp=tmp, 

153 state=gws.JobState.open, 

154 ) 

155 job = self._get_job_or_fail(job_uid, state=tmp) 

156 self._db().execute( 

157 f'UPDATE {self.TABLE} SET state=:s WHERE uid=:uid', 

158 uid=job.uid, 

159 s=gws.JobState.running, 

160 ) 

161 job = self._get_job_or_fail(job_uid, state=gws.JobState.running) 

162 

163 # now it's ours, let's run it 

164 

165 try: 

166 mod_path, _, fn_name = job.worker.rpartition(':') 

167 mod = gws.lib.dynimport.import_from_path(mod_path) 

168 worker_cls = getattr(mod, fn_name) 

169 worker_cls.run(self.root, job) 

170 except gws.JobTerminated as exc: 

171 gws.log.error(f'JOB {job_uid}: JobTerminated: {exc.args[0]!r}') 

172 except Exception as exc: 

173 gws.log.error(f'JOB {job_uid}: FAILED {exc=}') 

174 gws.log.exception() 

175 self.update_job(job, state=gws.JobState.error, error=repr(exc)) 

176 

177 return self.get_job(job_uid) 

178 

179 def cancel_job(self, job: gws.Job) -> Optional[gws.Job]: 

180 return self.update_job(job, state=gws.JobState.cancel) 

181 

182 def remove_job(self, job: gws.Job): 

183 self._db().delete(self.TABLE, job.uid) 

184 

185 def require_job(self, req, p): 

186 job = self.root.app.jobMgr.get_job(p.jobUid, req.user) 

187 if not job: 

188 raise gws.NotFoundError(f'JOB {p.jobUid}: not found') 

189 return job 

190 

191 def require_result(self, req, p): 

192 job = self.require_job(req, p) 

193 if job.state != gws.JobState.complete: 

194 raise gws.NotFoundError(f'JOB {p.jobUid}: wrong state {job.state!r}') 

195 if not job.result: 

196 raise gws.NotFoundError(f'JOB {p.jobUid}: no result') 

197 return job.result 

198 

199 def handle_status_request(self, req, p): 

200 job = self.require_job(req, p) 

201 return self.job_status_response(job) 

202 

203 def handle_cancel_request(self, req, p): 

204 job = self.require_job(req, p) 

205 job = self.cancel_job(job) 

206 if not job: 

207 raise gws.NotFoundError(f'JOB {p.jobUid}: not found') 

208 return self.job_status_response(job) 

209 

210 def job_status_response(self, job, **kwargs): 

211 d = dict( 

212 jobUid=job.uid, 

213 state=job.state, 

214 stepName=job.stepName or '', 

215 progress=self._get_progress(job), 

216 output={}, 

217 ) 

218 d.update(kwargs) 

219 return gws.JobStatusResponse(d) 

220 

221 def _get_progress(self, job): 

222 """Return the progress of a job in percent.""" 

223 if job.state == gws.JobState.complete: 

224 return 100 

225 if job.state != gws.JobState.running: 

226 return 0 

227 if not job.numSteps: 

228 return 0 

229 return int(min(100.0, job.step * 100.0 / job.numSteps)) 

230 

231 ## 

232 

233 _sqlitex: gws.lib.sqlitex.Object 

234 """Database object, created on first use.""" 

235 

236 def _db(self): 

237 """Return the database object, creating it if needed.""" 

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

239 self._sqlitex = gws.lib.sqlitex.Object(self.dbPath, self.DDL) 

240 return self._sqlitex