meerschaum.jobs

 50class Job:
 51    """
 52    Manage a `meerschaum.utils.daemon.Daemon`, locally or remotely via the API.
 53    """
 54
 55    def __init__(
 56        self,
 57        name: str,
 58        sysargs: Union[List[str], str, None] = None,
 59        env: Optional[Dict[str, str]] = None,
 60        executor_keys: Optional[str] = None,
 61        delete_after_completion: bool = False,
 62        _properties: Optional[Dict[str, Any]] = None,
 63        _rotating_log=None,
 64        _stdin_file=None,
 65        _status_hook: Optional[Callable[[], str]] = None,
 66        _result_hook: Optional[Callable[[], SuccessTuple]] = None,
 67        _externally_managed: bool = False,
 68    ):
 69        """
 70        Create a new job to manage a `meerschaum.utils.daemon.Daemon`.
 71
 72        Parameters
 73        ----------
 74        name: str
 75            The name of the job to be created.
 76            This will also be used as the Daemon ID.
 77
 78        sysargs: Union[List[str], str, None], default None
 79            The sysargs of the command to be executed, e.g. 'start api'.
 80
 81        env: Optional[Dict[str, str]], default None
 82            If provided, set these environment variables in the job's process.
 83
 84        executor_keys: Optional[str], default None
 85            If provided, execute the job remotely on an API instance, e.g. 'api:main'.
 86
 87        delete_after_completion: bool, default False
 88            If `True`, delete this job when it has finished executing.
 89
 90        _properties: Optional[Dict[str, Any]], default None
 91            If provided, use this to patch the daemon's properties.
 92        """
 93        from meerschaum.utils.daemon import Daemon
 94        for char in BANNED_CHARS:
 95            if char in name:
 96                raise ValueError(f"Invalid name: ({char}) is not allowed.")
 97
 98        if isinstance(sysargs, str):
 99            sysargs = shlex.split(sysargs)
100
101        and_key = STATIC_CONFIG['system']['arguments']['and_key']
102        escaped_and_key = STATIC_CONFIG['system']['arguments']['escaped_and_key']
103        if sysargs:
104            sysargs = [
105                (arg if arg != escaped_and_key else and_key)
106                for arg in sysargs
107            ]
108
109        ### NOTE: 'local' and 'systemd' executors are being coalesced.
110        if executor_keys is None:
111            from meerschaum.jobs import get_executor_keys_from_context
112            executor_keys = get_executor_keys_from_context()
113
114        self.executor_keys = executor_keys
115        self.name = name
116        try:
117            self._daemon = (
118                Daemon(daemon_id=name)
119                if executor_keys == 'local'
120                else None
121            )
122        except Exception:
123            self._daemon = None
124
125        ### Handle any injected dependencies.
126        if _rotating_log is not None:
127            self._rotating_log = _rotating_log
128            if self._daemon is not None:
129                self._daemon._rotating_log = _rotating_log
130
131        if _stdin_file is not None:
132            self._stdin_file = _stdin_file
133            if self._daemon is not None:
134                self._daemon._stdin_file = _stdin_file
135                self._daemon._blocking_stdin_file_path = _stdin_file.blocking_file_path
136
137        if _status_hook is not None:
138            self._status_hook = _status_hook
139
140        if _result_hook is not None:
141            self._result_hook = _result_hook
142
143        self._externally_managed = _externally_managed
144        self._properties_patch = _properties or {}
145        if _externally_managed:
146            self._properties_patch.update({'externally_managed': _externally_managed})
147
148        if env:
149            self._properties_patch.update({'env': env})
150
151        if delete_after_completion:
152            self._properties_patch.update({'delete_after_completion': delete_after_completion})
153
154        daemon_sysargs = (
155            self._daemon.properties.get('target', {}).get('args', [None])[0]
156            if self._daemon is not None
157            else None
158        )
159
160        if daemon_sysargs and sysargs and daemon_sysargs != sysargs:
161            warn("Given sysargs differ from existing sysargs.")
162
163        self._sysargs = [
164            arg
165            for arg in (daemon_sysargs or sysargs or [])
166            if arg not in ('-d', '--daemon')
167        ]
168        for restart_flag in RESTART_FLAGS:
169            if restart_flag in self._sysargs:
170                self._properties_patch.update({'restart': True})
171                break
172
173    @staticmethod
174    def from_pid(pid: int, executor_keys: Optional[str] = None) -> Job:
175        """
176        Build a `Job` from the PID of a running Meerschaum process.
177
178        Parameters
179        ----------
180        pid: int
181            The PID of the process.
182
183        executor_keys: Optional[str], default None
184            The executor keys to assign to the job.
185        """
186        from meerschaum.config.paths import DAEMON_RESOURCES_PATH
187
188        psutil = mrsm.attempt_import('psutil')
189        try:
190            process = psutil.Process(pid)
191        except psutil.NoSuchProcess as e:
192            warn(f"Process with PID {pid} does not exist.", stack=False)
193            raise e
194
195        command_args = process.cmdline()
196        is_daemon = command_args[1] == '-c'
197
198        if is_daemon:
199            daemon_id = command_args[-1].split('daemon_id=')[-1].split(')')[0].replace("'", '')
200            root_dir = process.environ().get(STATIC_CONFIG['environment']['root'], None)
201            if root_dir is None:
202                from meerschaum.config.paths import ROOT_DIR_PATH
203                root_dir = ROOT_DIR_PATH
204            else:
205                root_dir = pathlib.Path(root_dir)
206            jobs_dir = root_dir / DAEMON_RESOURCES_PATH.name
207            daemon_dir = jobs_dir / daemon_id
208            pid_file = daemon_dir / 'process.pid'
209
210            if pid_file.exists():
211                with open(pid_file, 'r', encoding='utf-8') as f:
212                    daemon_pid = int(f.read())
213
214                if pid != daemon_pid:
215                    raise EnvironmentError(f"Differing PIDs: {pid=}, {daemon_pid=}")
216            else:
217                raise EnvironmentError(f"Is job '{daemon_id}' running?")
218
219            return Job(daemon_id, executor_keys=executor_keys)
220
221        from meerschaum._internal.arguments._parse_arguments import parse_arguments
222        from meerschaum.utils.daemon import get_new_daemon_name
223
224        mrsm_ix = 0
225        for i, arg in enumerate(command_args):
226            if 'mrsm' in arg or 'meerschaum' in arg.lower():
227                mrsm_ix = i
228                break
229
230        sysargs = command_args[mrsm_ix+1:]
231        kwargs = parse_arguments(sysargs)
232        name = kwargs.get('name', get_new_daemon_name())
233        return Job(name, sysargs, executor_keys=executor_keys)
234
235    def start(self, debug: bool = False) -> SuccessTuple:
236        """
237        Start the job's daemon.
238        """
239        if self.executor is not None:
240            if not self.exists(debug=debug):
241                return self.executor.create_job(
242                    self.name,
243                    self.sysargs,
244                    properties=self.daemon.properties,
245                    debug=debug,
246                )
247            return self.executor.start_job(self.name, debug=debug)
248
249        if self.is_running():
250            return True, f"{self} is already running."
251
252        success, msg = self.daemon.run(
253            keep_daemon_output=(not self.delete_after_completion),
254            allow_dirty_run=True,
255        )
256        if not success:
257            return success, msg
258
259        return success, f"Started {self}."
260
261    def stop(self, timeout_seconds: Optional[int] = None, debug: bool = False) -> SuccessTuple:
262        """
263        Stop the job's daemon.
264        """
265        if self.executor is not None:
266            return self.executor.stop_job(self.name, debug=debug)
267
268        if self.daemon.status == 'stopped':
269            if not self.restart:
270                return True, f"{self} is not running."
271            elif self.stop_time is not None:
272                return True, f"{self} will not restart until manually started."
273
274        quit_success, quit_msg = self.daemon.quit(timeout=timeout_seconds)
275        if quit_success:
276            return quit_success, f"Stopped {self}."
277
278        warn(
279            f"Failed to gracefully quit {self}.",
280            stack=False,
281        )
282        kill_success, kill_msg = self.daemon.kill(timeout=timeout_seconds)
283        if not kill_success:
284            return kill_success, kill_msg
285
286        return kill_success, f"Killed {self}."
287
288    def pause(self, timeout_seconds: Optional[int] = None, debug: bool = False) -> SuccessTuple:
289        """
290        Pause the job's daemon.
291        """
292        if self.executor is not None:
293            return self.executor.pause_job(self.name, debug=debug)
294
295        pause_success, pause_msg = self.daemon.pause(timeout=timeout_seconds)
296        if not pause_success:
297            return pause_success, pause_msg
298
299        return pause_success, f"Paused {self}."
300
301    def delete(self, debug: bool = False) -> SuccessTuple:
302        """
303        Delete the job and its daemon.
304        """
305        if self.executor is not None:
306            return self.executor.delete_job(self.name, debug=debug)
307
308        if self.is_running():
309            stop_success, stop_msg = self.stop()
310            if not stop_success:
311                return stop_success, stop_msg
312
313        cleanup_success, cleanup_msg = self.daemon.cleanup()
314        if not cleanup_success:
315            return cleanup_success, cleanup_msg
316
317        _ = self.daemon._properties.pop('result', None)
318        return cleanup_success, f"Deleted {self}."
319
320    def is_running(self) -> bool:
321        """
322        Determine whether the job's daemon is running.
323        """
324        return self.status == 'running'
325
326    def exists(self, debug: bool = False) -> bool:
327        """
328        Determine whether the job exists.
329        """
330        if self.executor is not None:
331            return self.executor.get_job_exists(self.name, debug=debug)
332
333        return self.daemon.path.exists()
334
335    def get_logs(self) -> Union[str, None]:
336        """
337        Return the output text of the job's daemon.
338        """
339        if self.executor is not None:
340            return self.executor.get_logs(self.name)
341
342        return self.daemon.log_text
343
344    def monitor_logs(
345        self,
346        callback_function: Callable[[str], None] = partial(print, end=''),
347        input_callback_function: Optional[Callable[[], str]] = None,
348        stop_callback_function: Optional[Callable[[SuccessTuple], None]] = None,
349        stop_event: Optional[asyncio.Event] = None,
350        stop_on_exit: bool = False,
351        strip_timestamps: bool = False,
352        accept_input: bool = True,
353        debug: bool = False,
354    ):
355        """
356        Monitor the job's log files and execute a callback on new lines.
357
358        Parameters
359        ----------
360        callback_function: Callable[[str], None], default partial(print, end='')
361            The callback to execute as new data comes in.
362            Defaults to printing the output directly to `stdout`.
363
364        input_callback_function: Optional[Callable[[], str]], default None
365            If provided, execute this callback when the daemon is blocking on stdin.
366            Defaults to `sys.stdin.readline()`.
367
368        stop_callback_function: Optional[Callable[[SuccessTuple]], str], default None
369            If provided, execute this callback when the daemon stops.
370            The job's SuccessTuple will be passed to the callback.
371
372        stop_event: Optional[asyncio.Event], default None
373            If provided, stop monitoring when this event is set.
374            You may instead raise `meerschaum.jobs.StopMonitoringLogs`
375            from within `callback_function` to stop monitoring.
376
377        stop_on_exit: bool, default False
378            If `True`, stop monitoring when the job stops.
379
380        strip_timestamps: bool, default False
381            If `True`, remove leading timestamps from lines.
382
383        accept_input: bool, default True
384            If `True`, accept input when the daemon blocks on stdin.
385        """
386        def default_input_callback_function():
387            return sys.stdin.readline()
388
389        if input_callback_function is None:
390            input_callback_function = default_input_callback_function
391
392        if self.executor is not None:
393            self.executor.monitor_logs(
394                self.name,
395                callback_function,
396                input_callback_function=input_callback_function,
397                stop_callback_function=stop_callback_function,
398                stop_on_exit=stop_on_exit,
399                accept_input=accept_input,
400                strip_timestamps=strip_timestamps,
401                debug=debug,
402            )
403            return
404
405        monitor_logs_coroutine = self.monitor_logs_async(
406            callback_function=callback_function,
407            input_callback_function=input_callback_function,
408            stop_callback_function=stop_callback_function,
409            stop_event=stop_event,
410            stop_on_exit=stop_on_exit,
411            strip_timestamps=strip_timestamps,
412            accept_input=accept_input,
413        )
414        return asyncio.run(monitor_logs_coroutine)
415
416    async def monitor_logs_async(
417        self,
418        callback_function: Callable[[str], None] = partial(print, end='', flush=True),
419        input_callback_function: Optional[Callable[[], str]] = None,
420        stop_callback_function: Optional[Callable[[SuccessTuple], None]] = None,
421        stop_event: Optional[asyncio.Event] = None,
422        stop_on_exit: bool = False,
423        strip_timestamps: bool = False,
424        accept_input: bool = True,
425        _logs_path: Optional[pathlib.Path] = None,
426        _log=None,
427        _stdin_file=None,
428        debug: bool = False,
429    ):
430        """
431        Monitor the job's log files and await a callback on new lines.
432
433        Parameters
434        ----------
435        callback_function: Callable[[str], None], default partial(print, end='')
436            The callback to execute as new data comes in.
437            Defaults to printing the output directly to `stdout`.
438
439        input_callback_function: Optional[Callable[[], str]], default None
440            If provided, execute this callback when the daemon is blocking on stdin.
441            Defaults to `sys.stdin.readline()`.
442
443        stop_callback_function: Optional[Callable[[SuccessTuple]], str], default None
444            If provided, execute this callback when the daemon stops.
445            The job's SuccessTuple will be passed to the callback.
446
447        stop_event: Optional[asyncio.Event], default None
448            If provided, stop monitoring when this event is set.
449            You may instead raise `meerschaum.jobs.StopMonitoringLogs`
450            from within `callback_function` to stop monitoring.
451
452        stop_on_exit: bool, default False
453            If `True`, stop monitoring when the job stops.
454
455        strip_timestamps: bool, default False
456            If `True`, remove leading timestamps from lines.
457
458        accept_input: bool, default True
459            If `True`, accept input when the daemon blocks on stdin.
460        """
461        def default_input_callback_function():
462            return sys.stdin.readline()
463
464        if input_callback_function is None:
465            input_callback_function = default_input_callback_function
466
467        if self.executor is not None:
468            await self.executor.monitor_logs_async(
469                self.name,
470                callback_function,
471                input_callback_function=input_callback_function,
472                stop_callback_function=stop_callback_function,
473                stop_on_exit=stop_on_exit,
474                strip_timestamps=strip_timestamps,
475                accept_input=accept_input,
476                debug=debug,
477            )
478            return
479
480        from meerschaum.utils.formatting._jobs import strip_timestamp_from_line
481
482        events = {
483            'user': stop_event,
484            'stopped': asyncio.Event(),
485        }
486        combined_event = asyncio.Event()
487        emitted_text = False
488        stdin_file = _stdin_file if _stdin_file is not None else self.daemon.stdin_file
489
490        async def check_job_status():
491            nonlocal emitted_text
492            stopped_event = events.get('stopped', None)
493            if stopped_event is None:
494                return
495
496            sleep_time = 0.1
497            while sleep_time < 60:
498                if self.status == 'stopped':
499                    if not emitted_text:
500                        await asyncio.sleep(sleep_time)
501                        sleep_time = round(sleep_time * 1.1, 2)
502                        continue
503
504                    if stop_callback_function is not None:
505                        try:
506                            if asyncio.iscoroutinefunction(stop_callback_function):
507                                await stop_callback_function(self.result)
508                            else:
509                                stop_callback_function(self.result)
510                        except asyncio.exceptions.CancelledError:
511                            break
512                        except Exception:
513                            warn(traceback.format_exc())
514
515                    if stop_on_exit:
516                        events['stopped'].set()
517
518                    break
519                await asyncio.sleep(0.1)
520
521        async def check_blocking_on_input():
522            while True:
523                if not emitted_text or not self.is_blocking_on_stdin():
524                    try:
525                        await asyncio.sleep(0.1)
526                    except asyncio.exceptions.CancelledError:
527                        break
528                    continue
529
530                if not self.is_running():
531                    break
532
533                await emit_latest_lines()
534
535                try:
536                    print('', end='', flush=True)
537                    if asyncio.iscoroutinefunction(input_callback_function):
538                        data = await input_callback_function()
539                    else:
540                        data = input_callback_function()
541                except KeyboardInterrupt:
542                    break
543                if not data.endswith('\n'):
544                    data += '\n'
545
546                stdin_file.write(data)
547                await asyncio.sleep(0.1)
548
549        async def combine_events():
550            event_tasks = [
551                asyncio.create_task(event.wait())
552                for event in events.values()
553                if event is not None
554            ]
555            if not event_tasks:
556                return
557
558            try:
559                done, pending = await asyncio.wait(
560                    event_tasks,
561                    return_when=asyncio.FIRST_COMPLETED,
562                )
563                for task in pending:
564                    task.cancel()
565            except asyncio.exceptions.CancelledError:
566                pass
567            finally:
568                combined_event.set()
569
570        check_job_status_task = asyncio.create_task(check_job_status())
571        check_blocking_on_input_task = asyncio.create_task(check_blocking_on_input())
572        combine_events_task = asyncio.create_task(combine_events())
573
574        log = _log if _log is not None else self.daemon.rotating_log
575        lines_to_show = get_config('jobs', 'logs', 'lines_to_show')
576
577        async def emit_latest_lines():
578            nonlocal emitted_text
579            lines = log.readlines()
580            for line in lines[(-1 * lines_to_show):]:
581                if stop_event is not None and stop_event.is_set():
582                    return
583
584                if strip_timestamps:
585                    line = strip_timestamp_from_line(line)
586
587                try:
588                    if asyncio.iscoroutinefunction(callback_function):
589                        await callback_function(line)
590                    else:
591                        callback_function(line)
592                    emitted_text = True
593                except StopMonitoringLogs:
594                    return
595                except Exception:
596                    warn(f"Error in logs callback:\n{traceback.format_exc()}")
597
598        await emit_latest_lines()
599
600        tasks = (
601            [check_job_status_task]
602            + ([check_blocking_on_input_task] if accept_input else [])
603            + [combine_events_task]
604        )
605        try:
606            _ = asyncio.gather(*tasks, return_exceptions=True)
607        except asyncio.exceptions.CancelledError:
608            raise
609        except Exception:
610            warn(f"Failed to run async checks:\n{traceback.format_exc()}")
611
612        watchfiles = mrsm.attempt_import('watchfiles')
613        async for changes in watchfiles.awatch(
614            _logs_path or LOGS_RESOURCES_PATH,
615            stop_event=combined_event,
616        ):
617            for change in changes:
618                file_path_str = change[1]
619                file_path = pathlib.Path(file_path_str)
620                latest_subfile_path = log.get_latest_subfile_path()
621                if latest_subfile_path != file_path:
622                    continue
623
624                await emit_latest_lines()
625
626        await emit_latest_lines()
627
628    def is_blocking_on_stdin(self, debug: bool = False) -> bool:
629        """
630        Return whether a job's daemon is blocking on stdin.
631        """
632        if self.executor is not None:
633            return self.executor.get_job_is_blocking_on_stdin(self.name, debug=debug)
634
635        return self.is_running() and self.daemon.blocking_stdin_file_path.exists()
636
637    def write_stdin(self, data):
638        """
639        Write to a job's daemon's `stdin`.
640        """
641        self.daemon.stdin_file.write(data)
642
643    @property
644    def executor(self) -> Union[Executor, None]:
645        """
646        If the job is remote, return the connector to the remote API instance.
647        """
648        return (
649            mrsm.get_connector(self.executor_keys)
650            if self.executor_keys != 'local'
651            else None
652        )
653
654    @property
655    def status(self) -> str:
656        """
657        Return the running status of the job's daemon.
658        """
659        if '_status_hook' in self.__dict__:
660            return self._status_hook()
661
662        if self.executor is not None:
663            return self.executor.get_job_status(self.name)
664
665        return self.daemon.status
666
667    @property
668    def pid(self) -> Union[int, None]:
669        """
670        Return the PID of the job's dameon.
671        """
672        if self.executor is not None:
673            return self.executor.get_job_metadata(self.name).get('daemon', {}).get('pid', None)
674
675        return self.daemon.pid
676
677    @property
678    def restart(self) -> bool:
679        """
680        Return whether to restart a stopped job.
681        """
682        if self.executor is not None:
683            return self.executor.get_job_metadata(self.name).get('restart', False)
684
685        return self.daemon.properties.get('restart', False)
686
687    @property
688    def result(self) -> SuccessTuple:
689        """
690        Return the `SuccessTuple` when the job has terminated.
691        """
692        if self.is_running():
693            return True, f"{self} is running."
694
695        if '_result_hook' in self.__dict__:
696            return self._result_hook()
697
698        if self.executor is not None:
699            return (
700                self.executor.get_job_metadata(self.name)
701                .get('result', (False, "No result available."))
702            )
703
704        _result = self.daemon.properties.get('result', None)
705        if _result is None:
706            return False, "No result available."
707
708        return tuple(_result)
709
710    @property
711    def sysargs(self) -> List[str]:
712        """
713        Return the sysargs to use for the Daemon.
714        """
715        if self._sysargs:
716            return self._sysargs
717
718        if self.executor is not None:
719            return self.executor.get_job_metadata(self.name).get('sysargs', [])
720
721        target_args = self.daemon.target_args
722        if target_args is None:
723            return []
724        self._sysargs = target_args[0] if len(target_args) > 0 else []
725        return self._sysargs
726
727    @property
728    def daemon(self) -> 'Daemon':
729        """
730        Return the daemon which this job manages.
731        """
732        from meerschaum.utils.daemon import Daemon
733        if self._daemon is not None and self.executor is None and self._sysargs:
734            return self._daemon
735
736        remote_properties = (
737            {}
738            if self.executor is None
739            else self.executor.get_job_properties(self.name)
740        )
741        properties = {**remote_properties, **self._properties_patch}
742
743        self._daemon = Daemon(
744            target=entry,
745            target_args=[self._sysargs],
746            target_kw={},
747            daemon_id=self.name,
748            label=shlex.join(self._sysargs),
749            properties=properties,
750        )
751        if '_rotating_log' in self.__dict__:
752            self._daemon._rotating_log = self._rotating_log
753
754        if '_stdin_file' in self.__dict__:
755            self._daemon._stdin_file = self._stdin_file
756            self._daemon._blocking_stdin_file_path = self._stdin_file.blocking_file_path
757
758        return self._daemon
759
760    @property
761    def began(self) -> Union[datetime, None]:
762        """
763        The datetime when the job began running.
764        """
765        if self.executor is not None:
766            began_str = self.executor.get_job_began(self.name)
767            if began_str is None:
768                return None
769            return (
770                datetime.fromisoformat(began_str)
771                .astimezone(timezone.utc)
772                .replace(tzinfo=None)
773            )
774
775        began_str = self.daemon.properties.get('process', {}).get('began', None)
776        if began_str is None:
777            return None
778
779        return datetime.fromisoformat(began_str)
780
781    @property
782    def ended(self) -> Union[datetime, None]:
783        """
784        The datetime when the job stopped running.
785        """
786        if self.executor is not None:
787            ended_str = self.executor.get_job_ended(self.name)
788            if ended_str is None:
789                return None
790            return (
791                datetime.fromisoformat(ended_str)
792                .astimezone(timezone.utc)
793                .replace(tzinfo=None)
794            )
795
796        ended_str = self.daemon.properties.get('process', {}).get('ended', None)
797        if ended_str is None:
798            return None
799
800        return datetime.fromisoformat(ended_str)
801
802    @property
803    def paused(self) -> Union[datetime, None]:
804        """
805        The datetime when the job was suspended while running.
806        """
807        if self.executor is not None:
808            paused_str = self.executor.get_job_paused(self.name)
809            if paused_str is None:
810                return None
811            return (
812                datetime.fromisoformat(paused_str)
813                .astimezone(timezone.utc)
814                .replace(tzinfo=None)
815            )
816
817        paused_str = self.daemon.properties.get('process', {}).get('paused', None)
818        if paused_str is None:
819            return None
820
821        return datetime.fromisoformat(paused_str)
822
823    @property
824    def stop_time(self) -> Union[datetime, None]:
825        """
826        Return the timestamp when the job was manually stopped.
827        """
828        if self.executor is not None:
829            return self.executor.get_job_stop_time(self.name)
830
831        if not self.daemon.stop_path.exists():
832            return None
833
834        stop_data = self.daemon._read_stop_file()
835        if not stop_data:
836            return None
837
838        stop_time_str = stop_data.get('stop_time', None)
839        if not stop_time_str:
840            warn(f"Could not read stop time for {self}.")
841            return None
842
843        return datetime.fromisoformat(stop_time_str)
844
845    @property
846    def hidden(self) -> bool:
847        """
848        Return a bool indicating whether this job should be displayed.
849        """
850        return (
851            self.name.startswith('_')
852            or self.name.startswith('.')
853            or self._is_externally_managed
854        )
855
856    def check_restart(self) -> SuccessTuple:
857        """
858        If `restart` is `True` and the daemon is not running,
859        restart the job.
860        Do not restart if the job was manually stopped.
861        """
862        if self.is_running():
863            return True, f"{self} is running."
864
865        if not self.restart:
866            return True, f"{self} does not need to be restarted."
867
868        if self.stop_time is not None:
869            return True, f"{self} was manually stopped."
870
871        return self.start()
872
873    @property
874    def label(self) -> str:
875        """
876        Return the job's Daemon label (joined sysargs).
877        """
878        from meerschaum._internal.arguments import compress_pipeline_sysargs
879        sysargs = compress_pipeline_sysargs(self.sysargs)
880        return shlex.join(sysargs).replace(' + ', '\n+ ').replace(' : ', '\n: ').lstrip().rstrip()
881
882    @property
883    def _externally_managed_file(self) -> pathlib.Path:
884        """
885        Return the path to the externally managed file.
886        """
887        return self.daemon.path / '.externally-managed'
888
889    def _set_externally_managed(self):
890        """
891        Set this job as externally managed.
892        """
893        self._externally_managed = True
894        try:
895            self._externally_managed_file.parent.mkdir(exist_ok=True, parents=True)
896            self._externally_managed_file.touch()
897        except Exception as e:
898            warn(e)
899
900    @property
901    def _is_externally_managed(self) -> bool:
902        """
903        Return whether this job is externally managed.
904        """
905        return self.executor_keys in (None, 'local') and (
906            self._externally_managed or self._externally_managed_file.exists()
907        )
908
909    @property
910    def env(self) -> Dict[str, str]:
911        """
912        Return the environment variables to set for the job's process.
913        """
914        if '_env' in self.__dict__:
915            return self.__dict__['_env']
916
917        _env = self.daemon.properties.get('env', {})
918        default_env = {
919            'PYTHONUNBUFFERED': '1',
920            'LINES': str(get_config('jobs', 'terminal', 'lines')),
921            'COLUMNS': str(get_config('jobs', 'terminal', 'columns')),
922            STATIC_CONFIG['environment']['noninteractive']: 'true',
923        }
924        self._env = {**default_env, **_env}
925        return self._env
926
927    @property
928    def delete_after_completion(self) -> bool:
929        """
930        Return whether this job is configured to delete itself after completion.
931        """
932        if '_delete_after_completion' in self.__dict__:
933            return self.__dict__.get('_delete_after_completion', False)
934
935        self._delete_after_completion = self.daemon.properties.get('delete_after_completion', False)
936        return self._delete_after_completion
937
938    def __str__(self) -> str:
939        sysargs = self.sysargs
940        sysargs_str = shlex.join(sysargs) if sysargs else ''
941        job_str = f'Job("{self.name}"'
942        if sysargs_str:
943            job_str += f', "{sysargs_str}"'
944
945        job_str += ')'
946        return job_str
947
948    def __repr__(self) -> str:
949        return str(self)
950
951    def __hash__(self) -> int:
952        return hash(self.name)