meerschaum.utils.daemon.Daemon

Manage a job's daemon: its directory, its process and its logs.

Daemon is a read and control façade over the native daemon (crate meerschaum-jobs, bound as meerschaum._core.Daemon). A daemon runs a Meerschaum command (Daemon(sysargs=[...]), what Job manages); running an arbitrary Python callable as a daemon (Daemon(target=...)) is not supported in Meerschaum v5.

  1#! /usr/bin/env python3
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Manage a job's daemon: its directory, its process and its logs.
  7
  8`Daemon` is a read and control façade over the native daemon (crate `meerschaum-jobs`, bound as
  9`meerschaum._core.Daemon`). A daemon runs a Meerschaum command (`Daemon(sysargs=[...])`, what
 10`Job` manages); running an arbitrary Python callable as a daemon (`Daemon(target=...)`) is not
 11supported in Meerschaum v5.
 12"""
 13
 14from __future__ import annotations
 15
 16import pathlib
 17import shlex
 18
 19from meerschaum.utils.typing import Any, Callable, Dict, List, Optional, SuccessTuple, Union
 20
 21_REMOVED = (
 22    "Running a Python callable as a daemon is not supported in Meerschaum v5. "
 23    "Use `meerschaum.jobs.Job(name, sysargs)` to run a Meerschaum command in the background "
 24    "(see the migration guide, 'Removed in 5.0')."
 25)
 26
 27#: `Daemon.run()` results of this process, by daemon ID (kept for v4's `_results`).
 28_results: Dict[str, SuccessTuple] = {}
 29
 30
 31class Daemon:
 32    """
 33    Daemonize Python functions into background processes (v5: Meerschaum commands).
 34    """
 35
 36    def __init__(
 37        self,
 38        target: Optional[Callable[[Any], Any]] = None,
 39        target_args: Optional[List[Any]] = None,
 40        target_kw: Optional[Dict[str, Any]] = None,
 41        env: Optional[Dict[str, str]] = None,
 42        daemon_id: Optional[str] = None,
 43        label: Optional[str] = None,
 44        properties: Optional[Dict[str, Any]] = None,
 45        pickle: bool = True,
 46        *,
 47        sysargs: Optional[List[str]] = None,
 48    ):
 49        """
 50        Open an existing daemon (`Daemon(daemon_id='name')`) or describe a new one
 51        (`Daemon(daemon_id='name', sysargs=[...])`).
 52
 53        Parameters
 54        ----------
 55        target: Optional[Callable[[Any], Any]], default None
 56            Only `meerschaum._internal.entry.entry` (a Meerschaum command, with the sysargs as
 57            its first argument) is supported.
 58
 59        target_args: Optional[List[Any]], default None
 60            The arguments of `target`: `[sysargs]`.
 61
 62        env: Optional[Dict[str, str]], default None
 63            Environment variables to set in the daemon's process.
 64
 65        daemon_id: Optional[str], default None
 66            The name of the daemon (and of the job).
 67
 68        label: Optional[str], default None
 69            The label to show (the joined sysargs).
 70
 71        properties: Optional[Dict[str, Any]], default None
 72            Patch the daemon's properties.
 73        """
 74        from meerschaum.jobs import _native
 75        core = _native.core()
 76        if target is not None:
 77            if getattr(target, '__name__', None) != 'entry':
 78                raise NotImplementedError(_REMOVED)
 79            sysargs = list((target_args or [[]])[0] or [])
 80
 81        if target_kw:
 82            raise NotImplementedError(_REMOVED)
 83
 84        if daemon_id is None:
 85            if sysargs is None:
 86                raise ValueError("Pass a `daemon_id` to open an existing daemon.")
 87            daemon_id = core.jobs_new_daemon_name()
 88
 89        if env:
 90            properties = {**(properties or {}), 'env': dict(env)}
 91
 92        if sysargs is not None and label is None:
 93            label = shlex.join(sysargs)
 94
 95        self.daemon_id = daemon_id
 96        self._core = core.Daemon(daemon_id, sysargs, label, properties)
 97
 98    @classmethod
 99    def from_properties_file(cls, daemon_id: str) -> 'Daemon':
100        """Open the daemon `daemon_id` from its `properties.json`."""
101        return cls(daemon_id=daemon_id)
102
103    @property
104    def label(self) -> str:
105        return self._core.label
106
107    @property
108    def target(self) -> None:
109        return None
110
111    @property
112    def target_args(self) -> List[List[str]]:
113        """The arguments of the daemon's target: `[sysargs]`."""
114        return [self._core.sysargs]
115
116    @property
117    def target_kw(self) -> Dict[str, Any]:
118        return {}
119
120    @property
121    def hidden(self) -> bool:
122        return self._core.hidden
123
124    @property
125    def status(self) -> str:
126        """Return `'running'`, `'paused'` or `'stopped'`."""
127        return self._core.status
128
129    @property
130    def pid(self) -> Union[int, None]:
131        """Return the PID of the daemon's process, if it has one."""
132        return self._core.pid
133
134    @property
135    def process(self) -> Union[int, None]:
136        """Return the PID if the process exists (v4 returned a `psutil.Process`)."""
137        return self._core.process
138
139    @property
140    def properties(self) -> Dict[str, Any]:
141        """Return the daemon's properties (`properties.json` patched with this object's)."""
142        return self._core.properties()
143
144    @properties.setter
145    def properties(self, patch: Dict[str, Any]) -> None:
146        self._core.update_properties(patch)
147
148    def read_properties(self) -> Optional[Dict[str, Any]]:
149        """Read `properties.json`."""
150        return self._core.read_properties()
151
152    def write_properties(self) -> SuccessTuple:
153        """Write `properties.json`."""
154        return self._core.write_properties()
155
156    def _paths(self) -> Dict[str, str]:
157        return self._core.paths()
158
159    @property
160    def path(self) -> pathlib.Path:
161        """The daemon's directory (`jobs/<daemon_id>`)."""
162        return pathlib.Path(self._paths()['dir'])
163
164    @property
165    def properties_path(self) -> pathlib.Path:
166        return pathlib.Path(self._paths()['properties'])
167
168    @property
169    def pid_path(self) -> pathlib.Path:
170        return pathlib.Path(self._paths()['pid'])
171
172    @property
173    def pickle_path(self) -> pathlib.Path:
174        return pathlib.Path(self._paths()['pickle'])
175
176    @property
177    def stdin_file_path(self) -> pathlib.Path:
178        return pathlib.Path(self._paths()['stdin'])
179
180    @property
181    def blocking_stdin_file_path(self) -> pathlib.Path:
182        return pathlib.Path(self._paths()['stdin_block'])
183
184    @property
185    def prompt_kwargs_file_path(self) -> pathlib.Path:
186        return pathlib.Path(self._paths()['prompt_kwargs'])
187
188    @property
189    def stop_path(self) -> pathlib.Path:
190        return pathlib.Path(self._paths()['stop'])
191
192    @property
193    def log_path(self) -> pathlib.Path:
194        return pathlib.Path(self._paths()['log'])
195
196    @property
197    def log_offset_path(self) -> pathlib.Path:
198        return pathlib.Path(self._paths()['log_offset'])
199
200    @property
201    def stdin_file(self):
202        """A `StdinFile` for the daemon's stdin FIFO."""
203        from meerschaum.utils.daemon.StdinFile import StdinFile
204        return StdinFile(self.stdin_file_path, self.blocking_stdin_file_path)
205
206    @property
207    def rotating_log(self):
208        """A `RotatingFile` over the daemon's logs."""
209        from meerschaum.utils.daemon.RotatingFile import RotatingFile
210        logs = self.properties.get('logs', {}) or {}
211        return RotatingFile(
212            self.log_path,
213            num_files_to_keep=logs.get('num_files_to_keep'),
214            max_file_size=logs.get('max_file_size'),
215            write_timestamps=bool(logs.get('write_timestamps', False)),
216            timestamp_format=logs.get('timestamp_format'),
217        )
218
219    @property
220    def log_text(self) -> Optional[str]:
221        """The text of the daemon's logs."""
222        return self._core.log_text()
223
224    def readlines(self) -> List[str]:
225        """The log lines since the last call (the cursor is persisted)."""
226        return self._core.readlines()
227
228    def _read_stop_file(self) -> Optional[Dict[str, Any]]:
229        return self._core.read_stop_file()
230
231    def mkdir_if_not_exists(self, allow_dirty_run: bool = False) -> SuccessTuple:
232        """Create the daemon's directory."""
233        return self._core.mkdir_if_not_exists(allow_dirty_run)
234
235    def run(
236        self,
237        keep_daemon_output: bool = True,
238        allow_dirty_run: bool = False,
239        wait: bool = False,
240        timeout: Union[int, float] = 4,
241        debug: bool = False,
242    ) -> SuccessTuple:
243        """
244        Run the daemon's command in the background.
245
246        Parameters
247        ----------
248        keep_daemon_output: bool, default True
249            If `False`, delete the daemon's directory and logs when it finishes.
250
251        allow_dirty_run: bool, default False
252            If `True`, run although the daemon's directory exists.
253
254        wait: bool, default False
255            If `True`, wait up to `timeout` seconds until the daemon's process is running.
256
257        timeout: Union[int, float], default 4
258            How long to wait when `wait` is `True`.
259        """
260        result = self._core.run(keep_daemon_output, allow_dirty_run)
261        if wait and result[0]:
262            import time
263            deadline = time.monotonic() + timeout
264            while time.monotonic() < deadline and self.status == 'stopped':
265                time.sleep(0.05)
266            if self.status == 'stopped' and 'result' not in (self.read_properties() or {}):
267                result = False, f"Daemon '{self.daemon_id}' did not start within {timeout} seconds."
268        _results[self.daemon_id] = result
269        return result
270
271    def kill(
272        self,
273        timeout: Union[int, float, None] = 8,
274        debug: bool = False,
275    ) -> SuccessTuple:
276        """Forcibly terminate the daemon."""
277        return self._core.kill(timeout)
278
279    def quit(
280        self,
281        timeout: Union[int, float, None] = None,
282        debug: bool = False,
283    ) -> SuccessTuple:
284        """Gracefully quit the daemon (`SIGINT`)."""
285        return self._core.quit(timeout)
286
287    def pause(
288        self,
289        timeout: Union[int, float, None] = None,
290        check_timeout_interval: Union[float, int, None] = None,
291        debug: bool = False,
292    ) -> SuccessTuple:
293        """Suspend the daemon's process."""
294        return self._core.pause(timeout, check_timeout_interval)
295
296    def resume(
297        self,
298        timeout: Union[int, float, None] = None,
299        check_timeout_interval: Union[float, int, None] = None,
300        debug: bool = False,
301    ) -> SuccessTuple:
302        """Resume the daemon's process."""
303        return self._core.resume(timeout, check_timeout_interval)
304
305    def cleanup(self, keep_logs: bool = False, debug: bool = False) -> SuccessTuple:
306        """Remove the daemon's directory (and its logs unless `keep_logs`)."""
307        return self._core.cleanup(keep_logs)
308
309    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
310        """Return `timeout`, or the configured `jobs:timeout_seconds`."""
311        return self._core.get_timeout_seconds(timeout)
312
313    def get_check_timeout_interval_seconds(
314        self,
315        check_timeout_interval: Union[int, float, None] = None,
316    ) -> Union[int, float]:
317        """Return the interval, or the configured `jobs:check_timeout_interval_seconds`."""
318        return self._core.get_check_timeout_interval_seconds(check_timeout_interval)
319
320    def __eq__(self, other) -> bool:
321        return isinstance(other, Daemon) and other.daemon_id == self.daemon_id
322
323    def __hash__(self) -> int:
324        return hash(self.daemon_id)
325
326    def __str__(self) -> str:
327        return self.daemon_id
328
329    def __repr__(self) -> str:
330        return str(self)
class Daemon: 
 32class Daemon:
 33    """
 34    Daemonize Python functions into background processes (v5: Meerschaum commands).
 35    """
 36
 37    def __init__(
 38        self,
 39        target: Optional[Callable[[Any], Any]] = None,
 40        target_args: Optional[List[Any]] = None,
 41        target_kw: Optional[Dict[str, Any]] = None,
 42        env: Optional[Dict[str, str]] = None,
 43        daemon_id: Optional[str] = None,
 44        label: Optional[str] = None,
 45        properties: Optional[Dict[str, Any]] = None,
 46        pickle: bool = True,
 47        *,
 48        sysargs: Optional[List[str]] = None,
 49    ):
 50        """
 51        Open an existing daemon (`Daemon(daemon_id='name')`) or describe a new one
 52        (`Daemon(daemon_id='name', sysargs=[...])`).
 53
 54        Parameters
 55        ----------
 56        target: Optional[Callable[[Any], Any]], default None
 57            Only `meerschaum._internal.entry.entry` (a Meerschaum command, with the sysargs as
 58            its first argument) is supported.
 59
 60        target_args: Optional[List[Any]], default None
 61            The arguments of `target`: `[sysargs]`.
 62
 63        env: Optional[Dict[str, str]], default None
 64            Environment variables to set in the daemon's process.
 65
 66        daemon_id: Optional[str], default None
 67            The name of the daemon (and of the job).
 68
 69        label: Optional[str], default None
 70            The label to show (the joined sysargs).
 71
 72        properties: Optional[Dict[str, Any]], default None
 73            Patch the daemon's properties.
 74        """
 75        from meerschaum.jobs import _native
 76        core = _native.core()
 77        if target is not None:
 78            if getattr(target, '__name__', None) != 'entry':
 79                raise NotImplementedError(_REMOVED)
 80            sysargs = list((target_args or [[]])[0] or [])
 81
 82        if target_kw:
 83            raise NotImplementedError(_REMOVED)
 84
 85        if daemon_id is None:
 86            if sysargs is None:
 87                raise ValueError("Pass a `daemon_id` to open an existing daemon.")
 88            daemon_id = core.jobs_new_daemon_name()
 89
 90        if env:
 91            properties = {**(properties or {}), 'env': dict(env)}
 92
 93        if sysargs is not None and label is None:
 94            label = shlex.join(sysargs)
 95
 96        self.daemon_id = daemon_id
 97        self._core = core.Daemon(daemon_id, sysargs, label, properties)
 98
 99    @classmethod
100    def from_properties_file(cls, daemon_id: str) -> 'Daemon':
101        """Open the daemon `daemon_id` from its `properties.json`."""
102        return cls(daemon_id=daemon_id)
103
104    @property
105    def label(self) -> str:
106        return self._core.label
107
108    @property
109    def target(self) -> None:
110        return None
111
112    @property
113    def target_args(self) -> List[List[str]]:
114        """The arguments of the daemon's target: `[sysargs]`."""
115        return [self._core.sysargs]
116
117    @property
118    def target_kw(self) -> Dict[str, Any]:
119        return {}
120
121    @property
122    def hidden(self) -> bool:
123        return self._core.hidden
124
125    @property
126    def status(self) -> str:
127        """Return `'running'`, `'paused'` or `'stopped'`."""
128        return self._core.status
129
130    @property
131    def pid(self) -> Union[int, None]:
132        """Return the PID of the daemon's process, if it has one."""
133        return self._core.pid
134
135    @property
136    def process(self) -> Union[int, None]:
137        """Return the PID if the process exists (v4 returned a `psutil.Process`)."""
138        return self._core.process
139
140    @property
141    def properties(self) -> Dict[str, Any]:
142        """Return the daemon's properties (`properties.json` patched with this object's)."""
143        return self._core.properties()
144
145    @properties.setter
146    def properties(self, patch: Dict[str, Any]) -> None:
147        self._core.update_properties(patch)
148
149    def read_properties(self) -> Optional[Dict[str, Any]]:
150        """Read `properties.json`."""
151        return self._core.read_properties()
152
153    def write_properties(self) -> SuccessTuple:
154        """Write `properties.json`."""
155        return self._core.write_properties()
156
157    def _paths(self) -> Dict[str, str]:
158        return self._core.paths()
159
160    @property
161    def path(self) -> pathlib.Path:
162        """The daemon's directory (`jobs/<daemon_id>`)."""
163        return pathlib.Path(self._paths()['dir'])
164
165    @property
166    def properties_path(self) -> pathlib.Path:
167        return pathlib.Path(self._paths()['properties'])
168
169    @property
170    def pid_path(self) -> pathlib.Path:
171        return pathlib.Path(self._paths()['pid'])
172
173    @property
174    def pickle_path(self) -> pathlib.Path:
175        return pathlib.Path(self._paths()['pickle'])
176
177    @property
178    def stdin_file_path(self) -> pathlib.Path:
179        return pathlib.Path(self._paths()['stdin'])
180
181    @property
182    def blocking_stdin_file_path(self) -> pathlib.Path:
183        return pathlib.Path(self._paths()['stdin_block'])
184
185    @property
186    def prompt_kwargs_file_path(self) -> pathlib.Path:
187        return pathlib.Path(self._paths()['prompt_kwargs'])
188
189    @property
190    def stop_path(self) -> pathlib.Path:
191        return pathlib.Path(self._paths()['stop'])
192
193    @property
194    def log_path(self) -> pathlib.Path:
195        return pathlib.Path(self._paths()['log'])
196
197    @property
198    def log_offset_path(self) -> pathlib.Path:
199        return pathlib.Path(self._paths()['log_offset'])
200
201    @property
202    def stdin_file(self):
203        """A `StdinFile` for the daemon's stdin FIFO."""
204        from meerschaum.utils.daemon.StdinFile import StdinFile
205        return StdinFile(self.stdin_file_path, self.blocking_stdin_file_path)
206
207    @property
208    def rotating_log(self):
209        """A `RotatingFile` over the daemon's logs."""
210        from meerschaum.utils.daemon.RotatingFile import RotatingFile
211        logs = self.properties.get('logs', {}) or {}
212        return RotatingFile(
213            self.log_path,
214            num_files_to_keep=logs.get('num_files_to_keep'),
215            max_file_size=logs.get('max_file_size'),
216            write_timestamps=bool(logs.get('write_timestamps', False)),
217            timestamp_format=logs.get('timestamp_format'),
218        )
219
220    @property
221    def log_text(self) -> Optional[str]:
222        """The text of the daemon's logs."""
223        return self._core.log_text()
224
225    def readlines(self) -> List[str]:
226        """The log lines since the last call (the cursor is persisted)."""
227        return self._core.readlines()
228
229    def _read_stop_file(self) -> Optional[Dict[str, Any]]:
230        return self._core.read_stop_file()
231
232    def mkdir_if_not_exists(self, allow_dirty_run: bool = False) -> SuccessTuple:
233        """Create the daemon's directory."""
234        return self._core.mkdir_if_not_exists(allow_dirty_run)
235
236    def run(
237        self,
238        keep_daemon_output: bool = True,
239        allow_dirty_run: bool = False,
240        wait: bool = False,
241        timeout: Union[int, float] = 4,
242        debug: bool = False,
243    ) -> SuccessTuple:
244        """
245        Run the daemon's command in the background.
246
247        Parameters
248        ----------
249        keep_daemon_output: bool, default True
250            If `False`, delete the daemon's directory and logs when it finishes.
251
252        allow_dirty_run: bool, default False
253            If `True`, run although the daemon's directory exists.
254
255        wait: bool, default False
256            If `True`, wait up to `timeout` seconds until the daemon's process is running.
257
258        timeout: Union[int, float], default 4
259            How long to wait when `wait` is `True`.
260        """
261        result = self._core.run(keep_daemon_output, allow_dirty_run)
262        if wait and result[0]:
263            import time
264            deadline = time.monotonic() + timeout
265            while time.monotonic() < deadline and self.status == 'stopped':
266                time.sleep(0.05)
267            if self.status == 'stopped' and 'result' not in (self.read_properties() or {}):
268                result = False, f"Daemon '{self.daemon_id}' did not start within {timeout} seconds."
269        _results[self.daemon_id] = result
270        return result
271
272    def kill(
273        self,
274        timeout: Union[int, float, None] = 8,
275        debug: bool = False,
276    ) -> SuccessTuple:
277        """Forcibly terminate the daemon."""
278        return self._core.kill(timeout)
279
280    def quit(
281        self,
282        timeout: Union[int, float, None] = None,
283        debug: bool = False,
284    ) -> SuccessTuple:
285        """Gracefully quit the daemon (`SIGINT`)."""
286        return self._core.quit(timeout)
287
288    def pause(
289        self,
290        timeout: Union[int, float, None] = None,
291        check_timeout_interval: Union[float, int, None] = None,
292        debug: bool = False,
293    ) -> SuccessTuple:
294        """Suspend the daemon's process."""
295        return self._core.pause(timeout, check_timeout_interval)
296
297    def resume(
298        self,
299        timeout: Union[int, float, None] = None,
300        check_timeout_interval: Union[float, int, None] = None,
301        debug: bool = False,
302    ) -> SuccessTuple:
303        """Resume the daemon's process."""
304        return self._core.resume(timeout, check_timeout_interval)
305
306    def cleanup(self, keep_logs: bool = False, debug: bool = False) -> SuccessTuple:
307        """Remove the daemon's directory (and its logs unless `keep_logs`)."""
308        return self._core.cleanup(keep_logs)
309
310    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
311        """Return `timeout`, or the configured `jobs:timeout_seconds`."""
312        return self._core.get_timeout_seconds(timeout)
313
314    def get_check_timeout_interval_seconds(
315        self,
316        check_timeout_interval: Union[int, float, None] = None,
317    ) -> Union[int, float]:
318        """Return the interval, or the configured `jobs:check_timeout_interval_seconds`."""
319        return self._core.get_check_timeout_interval_seconds(check_timeout_interval)
320
321    def __eq__(self, other) -> bool:
322        return isinstance(other, Daemon) and other.daemon_id == self.daemon_id
323
324    def __hash__(self) -> int:
325        return hash(self.daemon_id)
326
327    def __str__(self) -> str:
328        return self.daemon_id
329
330    def __repr__(self) -> str:
331        return str(self)

Daemonize Python functions into background processes (v5: Meerschaum commands).

Daemon( target: Callable[[Any], Any] | None = None, target_args: List[Any] | None = None, target_kw: Dict[str, Any] | None = None, env: Dict[str, str] | None = None, daemon_id: str | None = None, label: str | None = None, properties: Dict[str, Any] | None = None, pickle: bool = True, *, sysargs: List[str] | None = None) 
37    def __init__(
38        self,
39        target: Optional[Callable[[Any], Any]] = None,
40        target_args: Optional[List[Any]] = None,
41        target_kw: Optional[Dict[str, Any]] = None,
42        env: Optional[Dict[str, str]] = None,
43        daemon_id: Optional[str] = None,
44        label: Optional[str] = None,
45        properties: Optional[Dict[str, Any]] = None,
46        pickle: bool = True,
47        *,
48        sysargs: Optional[List[str]] = None,
49    ):
50        """
51        Open an existing daemon (`Daemon(daemon_id='name')`) or describe a new one
52        (`Daemon(daemon_id='name', sysargs=[...])`).
53
54        Parameters
55        ----------
56        target: Optional[Callable[[Any], Any]], default None
57            Only `meerschaum._internal.entry.entry` (a Meerschaum command, with the sysargs as
58            its first argument) is supported.
59
60        target_args: Optional[List[Any]], default None
61            The arguments of `target`: `[sysargs]`.
62
63        env: Optional[Dict[str, str]], default None
64            Environment variables to set in the daemon's process.
65
66        daemon_id: Optional[str], default None
67            The name of the daemon (and of the job).
68
69        label: Optional[str], default None
70            The label to show (the joined sysargs).
71
72        properties: Optional[Dict[str, Any]], default None
73            Patch the daemon's properties.
74        """
75        from meerschaum.jobs import _native
76        core = _native.core()
77        if target is not None:
78            if getattr(target, '__name__', None) != 'entry':
79                raise NotImplementedError(_REMOVED)
80            sysargs = list((target_args or [[]])[0] or [])
81
82        if target_kw:
83            raise NotImplementedError(_REMOVED)
84
85        if daemon_id is None:
86            if sysargs is None:
87                raise ValueError("Pass a `daemon_id` to open an existing daemon.")
88            daemon_id = core.jobs_new_daemon_name()
89
90        if env:
91            properties = {**(properties or {}), 'env': dict(env)}
92
93        if sysargs is not None and label is None:
94            label = shlex.join(sysargs)
95
96        self.daemon_id = daemon_id
97        self._core = core.Daemon(daemon_id, sysargs, label, properties)

Open an existing daemon (Daemon(daemon_id='name')) or describe a new one (Daemon(daemon_id='name', sysargs=[...])).

Parameters
  • target (Optional[Callable[[Any], Any]], default None): Only meerschaum._internal.entry.entry (a Meerschaum command, with the sysargs as its first argument) is supported.
  • target_args (Optional[List[Any]], default None): The arguments of target: [sysargs].
  • env (Optional[Dict[str, str]], default None): Environment variables to set in the daemon's process.
  • daemon_id (Optional[str], default None): The name of the daemon (and of the job).
  • label (Optional[str], default None): The label to show (the joined sysargs).
  • properties (Optional[Dict[str, Any]], default None): Patch the daemon's properties.
daemon_id
@classmethod
def from_properties_file(cls, daemon_id: str) -> Daemon: 
 99    @classmethod
100    def from_properties_file(cls, daemon_id: str) -> 'Daemon':
101        """Open the daemon `daemon_id` from its `properties.json`."""
102        return cls(daemon_id=daemon_id)

Open the daemon daemon_id from its properties.json.

label: str 
104    @property
105    def label(self) -> str:
106        return self._core.label
target: None 
108    @property
109    def target(self) -> None:
110        return None
target_args: List[List[str]] 
112    @property
113    def target_args(self) -> List[List[str]]:
114        """The arguments of the daemon's target: `[sysargs]`."""
115        return [self._core.sysargs]

The arguments of the daemon's target: [sysargs].

target_kw: Dict[str, Any] 
117    @property
118    def target_kw(self) -> Dict[str, Any]:
119        return {}
hidden: bool 
121    @property
122    def hidden(self) -> bool:
123        return self._core.hidden
status: str 
125    @property
126    def status(self) -> str:
127        """Return `'running'`, `'paused'` or `'stopped'`."""
128        return self._core.status

Return 'running', 'paused' or 'stopped'.

pid: int | None 
130    @property
131    def pid(self) -> Union[int, None]:
132        """Return the PID of the daemon's process, if it has one."""
133        return self._core.pid

Return the PID of the daemon's process, if it has one.

process: int | None 
135    @property
136    def process(self) -> Union[int, None]:
137        """Return the PID if the process exists (v4 returned a `psutil.Process`)."""
138        return self._core.process

Return the PID if the process exists (v4 returned a psutil.Process).

properties: Dict[str, Any] 
140    @property
141    def properties(self) -> Dict[str, Any]:
142        """Return the daemon's properties (`properties.json` patched with this object's)."""
143        return self._core.properties()

Return the daemon's properties (properties.json patched with this object's).

def read_properties(self) -> Dict[str, Any] | None: 
149    def read_properties(self) -> Optional[Dict[str, Any]]:
150        """Read `properties.json`."""
151        return self._core.read_properties()

Read properties.json.

def write_properties(self) -> Tuple[bool, str]: 
153    def write_properties(self) -> SuccessTuple:
154        """Write `properties.json`."""
155        return self._core.write_properties()

Write properties.json.

path: pathlib.Path 
160    @property
161    def path(self) -> pathlib.Path:
162        """The daemon's directory (`jobs/<daemon_id>`)."""
163        return pathlib.Path(self._paths()['dir'])

The daemon's directory (jobs/<daemon_id>).

properties_path: pathlib.Path 
165    @property
166    def properties_path(self) -> pathlib.Path:
167        return pathlib.Path(self._paths()['properties'])
pid_path: pathlib.Path 
169    @property
170    def pid_path(self) -> pathlib.Path:
171        return pathlib.Path(self._paths()['pid'])
pickle_path: pathlib.Path 
173    @property
174    def pickle_path(self) -> pathlib.Path:
175        return pathlib.Path(self._paths()['pickle'])
stdin_file_path: pathlib.Path 
177    @property
178    def stdin_file_path(self) -> pathlib.Path:
179        return pathlib.Path(self._paths()['stdin'])
blocking_stdin_file_path: pathlib.Path 
181    @property
182    def blocking_stdin_file_path(self) -> pathlib.Path:
183        return pathlib.Path(self._paths()['stdin_block'])
prompt_kwargs_file_path: pathlib.Path 
185    @property
186    def prompt_kwargs_file_path(self) -> pathlib.Path:
187        return pathlib.Path(self._paths()['prompt_kwargs'])
stop_path: pathlib.Path 
189    @property
190    def stop_path(self) -> pathlib.Path:
191        return pathlib.Path(self._paths()['stop'])
log_path: pathlib.Path 
193    @property
194    def log_path(self) -> pathlib.Path:
195        return pathlib.Path(self._paths()['log'])
log_offset_path: pathlib.Path 
197    @property
198    def log_offset_path(self) -> pathlib.Path:
199        return pathlib.Path(self._paths()['log_offset'])
stdin_file 
201    @property
202    def stdin_file(self):
203        """A `StdinFile` for the daemon's stdin FIFO."""
204        from meerschaum.utils.daemon.StdinFile import StdinFile
205        return StdinFile(self.stdin_file_path, self.blocking_stdin_file_path)

A StdinFile for the daemon's stdin FIFO.

rotating_log 
207    @property
208    def rotating_log(self):
209        """A `RotatingFile` over the daemon's logs."""
210        from meerschaum.utils.daemon.RotatingFile import RotatingFile
211        logs = self.properties.get('logs', {}) or {}
212        return RotatingFile(
213            self.log_path,
214            num_files_to_keep=logs.get('num_files_to_keep'),
215            max_file_size=logs.get('max_file_size'),
216            write_timestamps=bool(logs.get('write_timestamps', False)),
217            timestamp_format=logs.get('timestamp_format'),
218        )

A RotatingFile over the daemon's logs.

log_text: str | None 
220    @property
221    def log_text(self) -> Optional[str]:
222        """The text of the daemon's logs."""
223        return self._core.log_text()

The text of the daemon's logs.

def readlines(self) -> List[str]: 
225    def readlines(self) -> List[str]:
226        """The log lines since the last call (the cursor is persisted)."""
227        return self._core.readlines()

The log lines since the last call (the cursor is persisted).

def mkdir_if_not_exists(self, allow_dirty_run: bool = False) -> Tuple[bool, str]: 
232    def mkdir_if_not_exists(self, allow_dirty_run: bool = False) -> SuccessTuple:
233        """Create the daemon's directory."""
234        return self._core.mkdir_if_not_exists(allow_dirty_run)

Create the daemon's directory.

def run( self, keep_daemon_output: bool = True, allow_dirty_run: bool = False, wait: bool = False, timeout: int | float = 4, debug: bool = False) -> Tuple[bool, str]: 
236    def run(
237        self,
238        keep_daemon_output: bool = True,
239        allow_dirty_run: bool = False,
240        wait: bool = False,
241        timeout: Union[int, float] = 4,
242        debug: bool = False,
243    ) -> SuccessTuple:
244        """
245        Run the daemon's command in the background.
246
247        Parameters
248        ----------
249        keep_daemon_output: bool, default True
250            If `False`, delete the daemon's directory and logs when it finishes.
251
252        allow_dirty_run: bool, default False
253            If `True`, run although the daemon's directory exists.
254
255        wait: bool, default False
256            If `True`, wait up to `timeout` seconds until the daemon's process is running.
257
258        timeout: Union[int, float], default 4
259            How long to wait when `wait` is `True`.
260        """
261        result = self._core.run(keep_daemon_output, allow_dirty_run)
262        if wait and result[0]:
263            import time
264            deadline = time.monotonic() + timeout
265            while time.monotonic() < deadline and self.status == 'stopped':
266                time.sleep(0.05)
267            if self.status == 'stopped' and 'result' not in (self.read_properties() or {}):
268                result = False, f"Daemon '{self.daemon_id}' did not start within {timeout} seconds."
269        _results[self.daemon_id] = result
270        return result

Run the daemon's command in the background.

Parameters
  • keep_daemon_output (bool, default True): If False, delete the daemon's directory and logs when it finishes.
  • allow_dirty_run (bool, default False): If True, run although the daemon's directory exists.
  • wait (bool, default False): If True, wait up to timeout seconds until the daemon's process is running.
  • timeout (Union[int, float], default 4): How long to wait when wait is True.
def kill( self, timeout: int | float | None = 8, debug: bool = False) -> Tuple[bool, str]: 
272    def kill(
273        self,
274        timeout: Union[int, float, None] = 8,
275        debug: bool = False,
276    ) -> SuccessTuple:
277        """Forcibly terminate the daemon."""
278        return self._core.kill(timeout)

Forcibly terminate the daemon.

def quit( self, timeout: int | float | None = None, debug: bool = False) -> Tuple[bool, str]: 
280    def quit(
281        self,
282        timeout: Union[int, float, None] = None,
283        debug: bool = False,
284    ) -> SuccessTuple:
285        """Gracefully quit the daemon (`SIGINT`)."""
286        return self._core.quit(timeout)

Gracefully quit the daemon (SIGINT).

def pause( self, timeout: int | float | None = None, check_timeout_interval: float | int | None = None, debug: bool = False) -> Tuple[bool, str]: 
288    def pause(
289        self,
290        timeout: Union[int, float, None] = None,
291        check_timeout_interval: Union[float, int, None] = None,
292        debug: bool = False,
293    ) -> SuccessTuple:
294        """Suspend the daemon's process."""
295        return self._core.pause(timeout, check_timeout_interval)

Suspend the daemon's process.

def resume( self, timeout: int | float | None = None, check_timeout_interval: float | int | None = None, debug: bool = False) -> Tuple[bool, str]: 
297    def resume(
298        self,
299        timeout: Union[int, float, None] = None,
300        check_timeout_interval: Union[float, int, None] = None,
301        debug: bool = False,
302    ) -> SuccessTuple:
303        """Resume the daemon's process."""
304        return self._core.resume(timeout, check_timeout_interval)

Resume the daemon's process.

def cleanup(self, keep_logs: bool = False, debug: bool = False) -> Tuple[bool, str]: 
306    def cleanup(self, keep_logs: bool = False, debug: bool = False) -> SuccessTuple:
307        """Remove the daemon's directory (and its logs unless `keep_logs`)."""
308        return self._core.cleanup(keep_logs)

Remove the daemon's directory (and its logs unless keep_logs).

def get_timeout_seconds(self, timeout: int | float | None = None) -> int | float: 
310    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
311        """Return `timeout`, or the configured `jobs:timeout_seconds`."""
312        return self._core.get_timeout_seconds(timeout)

Return timeout, or the configured jobs:timeout_seconds.

def get_check_timeout_interval_seconds(self, check_timeout_interval: int | float | None = None) -> int | float: 
314    def get_check_timeout_interval_seconds(
315        self,
316        check_timeout_interval: Union[int, float, None] = None,
317    ) -> Union[int, float]:
318        """Return the interval, or the configured `jobs:check_timeout_interval_seconds`."""
319        return self._core.get_check_timeout_interval_seconds(check_timeout_interval)

Return the interval, or the configured jobs:check_timeout_interval_seconds.