meerschaum.jobs

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