1import logging
2
3from omegaml.runtimes.proxies.baseproxy import RuntimeProxyBase
4
5logger = logging.getLogger(__file__)
6
7
[docs]
8class OmegaJobProxy(RuntimeProxyBase):
9 """
10 proxy to a remote job in a celery worker
11
12 Usage:
13
14 .. code::
15
16 om = Omega()
17 # result is AsyncResult, use .get() to return it's result
18 result = om.runtime.job('foojob').run()
19 result.get()
20
21 # result is AsyncResult, use .get() to return it's result
22 result = om.runtime.job('foojob').schedule()
23 result.get()
24 """
25
26 def __init__(self, jobname, runtime=None):
27 super().__init__(jobname, runtime=runtime, store=runtime.omega.jobs)
28 self.jobname = jobname
29
[docs]
30 def run(self, timeout=None, **kwargs):
31 """
32 submit the job for immediate execution
33
34 Args:
35 timeout (int): optional, timeout in seconds
36 **kwargs: kwargs to CeleryTask.delay
37
38 Returns:
39 AsyncResult
40 """
41 job_run = self.runtime.task('omegaml.notebook.tasks.run_omegaml_job')
42 return job_run.delay(self.jobname, timeout=timeout, **kwargs)
43
[docs]
44 def schedule(self, **kwargs):
45 """
46 schedule the job for repeated execution
47
48 Args:
49 **kwargs: see OmegaJob.schedule()
50 """
51 job_run = self.runtime.task('omegaml.notebook.tasks.schedule_omegaml_job')
52 return job_run.delay(self.jobname, **kwargs)
53
54 @property
55 def _mixins(self):
56 return self.runtime.omega.defaults.OMEGA_JOBPROXY_MIXINS