Coverage for gws-app/gws/base/job/worker.py: 90%

1"""Base job worker.""" 

2 

3from typing import Optional 

4import gws 

5 

6 

7class Object: 

8 """Base job worker. 

9 

10 Subclasses implement a ``run(root, job)`` class method, which the job 

11 manager calls to run the job. A worker can also run without a job, then 

12 job updates are skipped. 

13 """ 

14 

15 jobUid: str 

16 """Uid of the job, empty if the worker runs without a job or the job was terminated.""" 

17 user: gws.User 

18 """User the job runs for.""" 

19 

20 def __init__(self, root: gws.Root, user: gws.User, job: Optional[gws.Job] = None): 

21 """Create a worker. 

22 

23 Args: 

24 root: Root object. 

25 user: User the job runs for. 

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

27 """ 

28 self.jobUid = job.uid if job else '' 

29 self.root = root 

30 self.user = user 

31 

32 def get_job(self) -> Optional[gws.Job]: 

33 """Return the current state of the job. 

34 

35 Returns: 

36 The job, or ``None`` if the worker runs without a job. 

37 

38 Raises: 

39 ``gws.JobTerminated``: If the job no longer exists or is not running, e.g. because it was cancelled. 

40 """ 

41 if not self.jobUid: 

42 return 

43 

44 job = self.root.app.jobMgr.get_job( 

45 self.jobUid, 

46 user=self.user, 

47 state=gws.JobState.running 

48 ) 

49 if not job: 

50 self.jobUid = '' 

51 raise gws.JobTerminated('JOB TERMINATED') 

52 return job 

53 

54 def update_job(self, **kwargs): 

55 """Update the job record. Does nothing if the worker runs without a job. 

56 

57 Args: 

58 **kwargs: Job fields to update, e.g. ``state``, ``step``, ``numSteps``, ``stepName`` or ``result``. 

59 

60 Raises: 

61 ``gws.JobTerminated``: If the job no longer exists or is not running. 

62 """ 

63 job = self.get_job() 

64 if job: 

65 self.root.app.jobMgr.update_job(job, **kwargs)