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