meerschaum.utils.daemon

Manage the daemons behind jobs (meerschaum.jobs.Job is the way to run one).

The daemon functions are façades over the native jobs (meerschaum-jobs). Running an arbitrary Python callable as a daemon (Daemon(target=...), run_daemon, daemon_action, daemon_entry) and the file-descriptor interception of v4 are removed in Meerschaum v5.

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Manage the daemons behind jobs (`meerschaum.jobs.Job` is the way to run one).
  7
  8The daemon functions are façades over the native jobs (`meerschaum-jobs`). Running an arbitrary
  9Python callable as a daemon (`Daemon(target=...)`, `run_daemon`, `daemon_action`, `daemon_entry`)
 10and the file-descriptor interception of v4 are removed in Meerschaum v5.
 11"""
 12
 13from __future__ import annotations
 14
 15import os
 16
 17from meerschaum.utils.typing import List, Optional, Union
 18from meerschaum.utils.daemon.StdinFile import StdinFile
 19from meerschaum.utils.daemon.Daemon import Daemon
 20from meerschaum.utils.daemon.RotatingFile import RotatingFile
 21from meerschaum.utils.daemon._names import get_new_daemon_name
 22
 23
 24__all__ = (
 25    'get_daemons',
 26    'get_daemon_ids',
 27    'get_running_daemons',
 28    'get_stopped_daemons',
 29    'get_paused_daemons',
 30    'get_filtered_daemons',
 31    'get_new_daemon_name',
 32    'get_current_daemon',
 33    'running_in_daemon',
 34    'Daemon',
 35    'StdinFile',
 36    'RotatingFile',
 37)
 38
 39_daemons = {}
 40
 41
 42def get_daemon_ids() -> List[str]:
 43    """
 44    Return the IDs of all daemons on disk.
 45    """
 46    from meerschaum.jobs import _native
 47    return _native.core().jobs_daemon_ids()
 48
 49
 50def get_daemons() -> List[Daemon]:
 51    """
 52    Return all existing Daemons, sorted by end time.
 53    """
 54    daemons = [Daemon(daemon_id=d_id) for d_id in get_daemon_ids()]
 55    statuses = {daemon: daemon.status for daemon in daemons}
 56    key_by_status = {'stopped': 'ended', 'paused': 'paused', 'running': 'began'}
 57
 58    def _sorted(status: str) -> List[Daemon]:
 59        key = key_by_status[status]
 60        selected = [daemon for daemon in daemons if statuses[daemon] == status]
 61        return sorted(
 62            selected,
 63            key=lambda daemon: daemon.properties.get('process', {}).get(key, '9999'),
 64        )
 65
 66    return _sorted('stopped') + _sorted('paused') + _sorted('running')
 67
 68
 69def get_running_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
 70    """
 71    Return a list of currently running daemons.
 72    """
 73    if daemons is None:
 74        daemons = get_daemons()
 75    return [d for d in daemons if d.status == 'running']
 76
 77
 78def get_paused_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
 79    """
 80    Return a list of active but paused daemons.
 81    """
 82    if daemons is None:
 83        daemons = get_daemons()
 84    return [d for d in daemons if d.status == 'paused']
 85
 86
 87def get_stopped_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
 88    """
 89    Return a list of stopped daemons.
 90    """
 91    if daemons is None:
 92        daemons = get_daemons()
 93    return [d for d in daemons if d.status == 'stopped']
 94
 95
 96def get_filtered_daemons(
 97    filter_list: Optional[List[str]] = None,
 98    warn: bool = False,
 99) -> List[Daemon]:
100    """
101    Return a list of `Daemons` filtered by a list of `daemon_ids`.
102    Only `Daemons` that exist are returned.
103
104    If `filter_list` is `None` or empty, return all `Daemons` (from `get_daemons()`).
105
106    Parameters
107    ----------
108    filter_list: Optional[List[str]], default None
109        List of `daemon_ids` to include. If `daemon_ids` is `None` or empty,
110        return all `Daemons`.
111
112    warn: bool, default False
113        If `True`, raise warnings for non-existent `daemon_ids`.
114
115    Returns
116    -------
117    A list of Daemon objects.
118    """
119    if not filter_list:
120        daemons = get_daemons()
121        return [d for d in daemons if not d.hidden]
122
123    from meerschaum.utils.warnings import warn as _warn
124    daemons = []
125    for d_id in filter_list:
126        try:
127            d = Daemon(daemon_id=d_id)
128            _exists = d.path.exists()
129        except Exception:
130            _exists = False
131        if not _exists:
132            if warn:
133                _warn(f"Daemon '{d_id}' does not exist.", stack=False)
134            continue
135        daemons.append(d)
136    return daemons
137
138
139def running_in_daemon() -> bool:
140    """
141    Return whether the current thread is running in a Daemon context.
142    """
143    from meerschaum._internal.static import STATIC_CONFIG
144    daemon_env_var = STATIC_CONFIG['environment']['daemon_id']
145    return daemon_env_var in os.environ
146
147
148def get_current_daemon() -> Union[Daemon, None]:
149    """
150    If running withing a daemon context, return the corresponding `Daemon`.
151    Otherwise return `None`.
152    """
153    from meerschaum._internal.static import STATIC_CONFIG
154    daemon_env_var = STATIC_CONFIG['environment']['daemon_id']
155    daemon_id = os.environ.get(daemon_env_var, None)
156    if daemon_id is None:
157        return None
158
159    if daemon_id in _daemons:
160        return _daemons[daemon_id]
161    try:
162        _daemons[daemon_id] = Daemon(daemon_id=daemon_id)
163    except Exception:
164        return None
165    return _daemons[daemon_id]
def get_daemons() -> List[Daemon]: 
51def get_daemons() -> List[Daemon]:
52    """
53    Return all existing Daemons, sorted by end time.
54    """
55    daemons = [Daemon(daemon_id=d_id) for d_id in get_daemon_ids()]
56    statuses = {daemon: daemon.status for daemon in daemons}
57    key_by_status = {'stopped': 'ended', 'paused': 'paused', 'running': 'began'}
58
59    def _sorted(status: str) -> List[Daemon]:
60        key = key_by_status[status]
61        selected = [daemon for daemon in daemons if statuses[daemon] == status]
62        return sorted(
63            selected,
64            key=lambda daemon: daemon.properties.get('process', {}).get(key, '9999'),
65        )
66
67    return _sorted('stopped') + _sorted('paused') + _sorted('running')

Return all existing Daemons, sorted by end time.

def get_daemon_ids() -> List[str]: 
43def get_daemon_ids() -> List[str]:
44    """
45    Return the IDs of all daemons on disk.
46    """
47    from meerschaum.jobs import _native
48    return _native.core().jobs_daemon_ids()

Return the IDs of all daemons on disk.

def get_running_daemons( daemons: List[Daemon] | None = None) -> List[Daemon]: 
70def get_running_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
71    """
72    Return a list of currently running daemons.
73    """
74    if daemons is None:
75        daemons = get_daemons()
76    return [d for d in daemons if d.status == 'running']

Return a list of currently running daemons.

def get_stopped_daemons( daemons: List[Daemon] | None = None) -> List[Daemon]: 
88def get_stopped_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
89    """
90    Return a list of stopped daemons.
91    """
92    if daemons is None:
93        daemons = get_daemons()
94    return [d for d in daemons if d.status == 'stopped']

Return a list of stopped daemons.

def get_paused_daemons( daemons: List[Daemon] | None = None) -> List[Daemon]: 
79def get_paused_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
80    """
81    Return a list of active but paused daemons.
82    """
83    if daemons is None:
84        daemons = get_daemons()
85    return [d for d in daemons if d.status == 'paused']

Return a list of active but paused daemons.

def get_filtered_daemons( filter_list: List[str] | None = None, warn: bool = False) -> List[Daemon]: 
 97def get_filtered_daemons(
 98    filter_list: Optional[List[str]] = None,
 99    warn: bool = False,
100) -> List[Daemon]:
101    """
102    Return a list of `Daemons` filtered by a list of `daemon_ids`.
103    Only `Daemons` that exist are returned.
104
105    If `filter_list` is `None` or empty, return all `Daemons` (from `get_daemons()`).
106
107    Parameters
108    ----------
109    filter_list: Optional[List[str]], default None
110        List of `daemon_ids` to include. If `daemon_ids` is `None` or empty,
111        return all `Daemons`.
112
113    warn: bool, default False
114        If `True`, raise warnings for non-existent `daemon_ids`.
115
116    Returns
117    -------
118    A list of Daemon objects.
119    """
120    if not filter_list:
121        daemons = get_daemons()
122        return [d for d in daemons if not d.hidden]
123
124    from meerschaum.utils.warnings import warn as _warn
125    daemons = []
126    for d_id in filter_list:
127        try:
128            d = Daemon(daemon_id=d_id)
129            _exists = d.path.exists()
130        except Exception:
131            _exists = False
132        if not _exists:
133            if warn:
134                _warn(f"Daemon '{d_id}' does not exist.", stack=False)
135            continue
136        daemons.append(d)
137    return daemons

Return a list of Daemons filtered by a list of daemon_ids. Only Daemons that exist are returned.

If filter_list is None or empty, return all Daemons (from get_daemons()).

Parameters
  • filter_list (Optional[List[str]], default None): List of daemon_ids to include. If daemon_ids is None or empty, return all Daemons.
  • warn (bool, default False): If True, raise warnings for non-existent daemon_ids.
Returns
  • A list of Daemon objects.
def get_new_daemon_name() -> str: 
20def get_new_daemon_name() -> str:
21    """
22    Return a random name which is not used by an existing job.
23    """
24    from meerschaum.jobs import _native
25    return _native.core().jobs_new_daemon_name()

Return a random name which is not used by an existing job.

def get_current_daemon() -> Daemon | None: 
149def get_current_daemon() -> Union[Daemon, None]:
150    """
151    If running withing a daemon context, return the corresponding `Daemon`.
152    Otherwise return `None`.
153    """
154    from meerschaum._internal.static import STATIC_CONFIG
155    daemon_env_var = STATIC_CONFIG['environment']['daemon_id']
156    daemon_id = os.environ.get(daemon_env_var, None)
157    if daemon_id is None:
158        return None
159
160    if daemon_id in _daemons:
161        return _daemons[daemon_id]
162    try:
163        _daemons[daemon_id] = Daemon(daemon_id=daemon_id)
164    except Exception:
165        return None
166    return _daemons[daemon_id]

If running withing a daemon context, return the corresponding Daemon. Otherwise return None.

def running_in_daemon() -> bool: 
140def running_in_daemon() -> bool:
141    """
142    Return whether the current thread is running in a Daemon context.
143    """
144    from meerschaum._internal.static import STATIC_CONFIG
145    daemon_env_var = STATIC_CONFIG['environment']['daemon_id']
146    return daemon_env_var in os.environ

Return whether the current thread is running in a Daemon context.

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.

class StdinFile(io.IOBase): 
18class StdinFile(io.IOBase):
19    """
20    The client end of a job's stdin FIFO: `write()` sends input to the job.
21    """
22
23    def __init__(
24        self,
25        file_path: Union[pathlib.Path, str],
26        lock_file_path: Optional[pathlib.Path] = None,
27        decode: bool = True,
28        refresh_seconds: Union[int, float, None] = None,
29    ):
30        from meerschaum.jobs import _native
31        self.file_path = pathlib.Path(file_path)
32        self.blocking_file_path = (
33            pathlib.Path(lock_file_path)
34            if lock_file_path is not None
35            else (self.file_path.parent / (self.file_path.name + '.block'))
36        )
37        self.decode = decode
38        self._refresh_seconds = refresh_seconds
39        self._core = _native.core().StdinFile(
40            str(self.file_path),
41            str(self.blocking_file_path),
42        )
43
44    @property
45    def encoding(self) -> str:
46        return 'utf-8'
47
48    @property
49    def is_open(self) -> bool:
50        return self._core.is_open()
51
52    def readable(self) -> bool:
53        return True
54
55    def writable(self) -> bool:
56        return True
57
58    def seekable(self) -> bool:
59        return False
60
61    def isatty(self) -> bool:
62        return False
63
64    def write(self, data: Union[str, bytes]) -> int:
65        """Send input to the job."""
66        text = data.decode('utf-8') if isinstance(data, bytes) else data
67        return self._core.write(text)
68
69    def writelines(self, lines) -> None:
70        for line in lines:
71            self.write(line)
72
73    def readline(self, *args, **kw) -> str:
74        """Read a line from the FIFO (job side)."""
75        return self._core.readline()
76
77    def close(self) -> None:
78        self._core.close()

The client end of a job's stdin FIFO: write() sends input to the job.

StdinFile( file_path: pathlib.Path | str, lock_file_path: pathlib.Path | None = None, decode: bool = True, refresh_seconds: int | float | None = None) 
23    def __init__(
24        self,
25        file_path: Union[pathlib.Path, str],
26        lock_file_path: Optional[pathlib.Path] = None,
27        decode: bool = True,
28        refresh_seconds: Union[int, float, None] = None,
29    ):
30        from meerschaum.jobs import _native
31        self.file_path = pathlib.Path(file_path)
32        self.blocking_file_path = (
33            pathlib.Path(lock_file_path)
34            if lock_file_path is not None
35            else (self.file_path.parent / (self.file_path.name + '.block'))
36        )
37        self.decode = decode
38        self._refresh_seconds = refresh_seconds
39        self._core = _native.core().StdinFile(
40            str(self.file_path),
41            str(self.blocking_file_path),
42        )
file_path
blocking_file_path
decode
encoding: str 
44    @property
45    def encoding(self) -> str:
46        return 'utf-8'
is_open: bool 
48    @property
49    def is_open(self) -> bool:
50        return self._core.is_open()
def readable(self) -> bool: 
52    def readable(self) -> bool:
53        return True

Return whether object was opened for reading.

If False, read() will raise OSError.

def writable(self) -> bool: 
55    def writable(self) -> bool:
56        return True

Return whether object was opened for writing.

If False, write() will raise OSError.

def seekable(self) -> bool: 
58    def seekable(self) -> bool:
59        return False

Return whether object supports random access.

If False, seek(), tell() and truncate() will raise OSError. This method may need to do a test seek().

def isatty(self) -> bool: 
61    def isatty(self) -> bool:
62        return False

Return whether this is an 'interactive' stream.

Return False if it can't be determined.

def write(self, data: str | bytes) -> int: 
64    def write(self, data: Union[str, bytes]) -> int:
65        """Send input to the job."""
66        text = data.decode('utf-8') if isinstance(data, bytes) else data
67        return self._core.write(text)

Send input to the job.

def writelines(self, lines) -> None: 
69    def writelines(self, lines) -> None:
70        for line in lines:
71            self.write(line)

Write a list of lines to stream.

Line separators are not added, so it is usual for each of the lines provided to have a line separator at the end.

def readline(self, *args, **kw) -> str: 
73    def readline(self, *args, **kw) -> str:
74        """Read a line from the FIFO (job side)."""
75        return self._core.readline()

Read a line from the FIFO (job side).

def close(self) -> None: 
77    def close(self) -> None:
78        self._core.close()

Flush and close the IO object.

This method has no effect if the file is already closed.

class RotatingFile(io.IOBase): 
 18class RotatingFile(io.IOBase):
 19    """
 20    A `RotatingFile` may be treated like a normal file-like object.
 21    Under the hood, however, it will create new sub-files and delete old ones.
 22
 23    The stream redirection of v4 (`redirect_streams`, `start_log_fd_interception`) is not
 24    available: the job supervisor copies a job's output into its log.
 25    """
 26
 27    SEEK_BACK_ATTEMPTS: int = 5
 28
 29    def __init__(
 30        self,
 31        file_path: pathlib.Path,
 32        num_files_to_keep: Optional[int] = None,
 33        max_file_size: Optional[int] = None,
 34        redirect_streams: bool = False,
 35        write_timestamps: bool = False,
 36        timestamp_format: Optional[str] = None,
 37        write_callback=None,
 38    ):
 39        """
 40        Create a file-like object which manages other files.
 41
 42        Parameters
 43        ----------
 44        file_path: pathlib.Path
 45            The base path (`logs/<name>.log`); subfiles are `<name>.log.<n>`.
 46
 47        num_files_to_keep: int, default None
 48            How many sub-files to keep at any given time.
 49            Defaults to the configured value (5).
 50
 51        max_file_size: int, default None
 52            How large in bytes each sub-file can grow before another file is created.
 53            Defaults to the configured value (100_000).
 54
 55        write_timestamps: bool, default False
 56            If `True`, prepend the current UTC timestamp to each line of the file.
 57
 58        timestamp_format: str, default None
 59            If `write_timestamps` is `True`, use this format for the timestamps.
 60            Defaults to `'%Y-%m-%d %H:%M'`.
 61        """
 62        from meerschaum.config import get_config
 63        from meerschaum.jobs import _native
 64        if redirect_streams or write_callback is not None:
 65            raise NotImplementedError(
 66                "`redirect_streams` and `write_callback` are not supported in Meerschaum v5 "
 67                "(see the migration guide, 'Removed in 5.0')."
 68            )
 69        self.file_path = pathlib.Path(file_path)
 70        if num_files_to_keep is None:
 71            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
 72        if max_file_size is None:
 73            max_file_size = get_config('jobs', 'logs', 'max_file_size')
 74        if timestamp_format is None:
 75            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
 76        if num_files_to_keep < 1:
 77            raise ValueError("At least 1 file must be kept.")
 78        if max_file_size < 100:
 79            raise ValueError("Subfiles must contain at least 100 bytes.")
 80        self.num_files_to_keep = num_files_to_keep
 81        self.max_file_size = max_file_size
 82        self.write_timestamps = write_timestamps
 83        self.timestamp_format = timestamp_format
 84        self._core = _native.core().RotatingFile(
 85            str(self.file_path),
 86            num_files_to_keep,
 87            max_file_size,
 88            write_timestamps,
 89            timestamp_format,
 90        )
 91        self._closed = False
 92        self._lines_buffer: List[str] = []
 93
 94    @property
 95    def closed(self) -> bool:
 96        return self._closed
 97
 98    def readable(self) -> bool:
 99        return True
100
101    def writable(self) -> bool:
102        return True
103
104    def seekable(self) -> bool:
105        return True
106
107    def isatty(self) -> bool:
108        return False
109
110    def fileno(self) -> int:
111        raise io.UnsupportedOperation("A RotatingFile has no file descriptor.")
112
113    def read(self, *args, **kw) -> str:
114        """Return the text of every subfile."""
115        return self._core.read()
116
117    def readlines(self, *args, **kw) -> List[str]:
118        """Return the lines since the cursor."""
119        return self._core.readlines()
120
121    def readline(self, *args, **kw) -> str:
122        """Return the next line since the cursor ('' at the end)."""
123        if not self._lines_buffer:
124            self._lines_buffer = self._core.readlines()
125        return self._lines_buffer.pop(0) if self._lines_buffer else ''
126
127    def write(self, data: str) -> int:
128        """Write text, rotating subfiles as they fill."""
129        return self._core.write(data)
130
131    def writelines(self, lines) -> None:
132        for line in lines:
133            self.write(line)
134
135    def seek(self, position: int, *args, **kw) -> int:
136        """Move the read cursor in the latest subfile."""
137        self._lines_buffer = []
138        self._core.seek(position)
139        return position
140
141    def touch(self) -> None:
142        """Create the latest subfile."""
143        self._core.touch()
144
145    def refresh_cursor(self) -> None:
146        self._core.refresh_cursor()
147
148    def flush(self) -> None:
149        pass
150
151    def close(self) -> None:
152        """Close the open subfile handles."""
153        self._core.close()
154        self._closed = True
155
156    def delete(self, unused_only: bool = False) -> None:
157        """Delete the subfiles (`unused_only`: only those beyond the files to keep)."""
158        self._core.delete(unused_only)
159
160    def get_latest_subfile_index(self) -> int:
161        return self._core.get_latest_subfile_index()
162
163    def get_latest_subfile_path(self) -> pathlib.Path:
164        return pathlib.Path(self._core.get_latest_subfile_path())
165
166    def get_existing_subfile_indices(self) -> List[int]:
167        return self._core.get_existing_subfile_indices()
168
169    def get_existing_subfile_paths(self) -> List[pathlib.Path]:
170        return [pathlib.Path(p) for p in self._core.get_existing_subfile_paths()]
171
172    def get_subfile_path_from_index(self, index: int) -> pathlib.Path:
173        return pathlib.Path(self._core.get_subfile_path_from_index(index))
174
175    def get_subfile_name_from_index(self, index: int) -> str:
176        return self._core.get_subfile_name_from_index(index)
177
178    def get_index_from_subfile_name(self, subfile_name: str) -> int:
179        return self._core.get_index_from_subfile_name(subfile_name)
180
181    def get_timestamp_prefix_str(self) -> str:
182        return self._core.get_timestamp_prefix_str()

A RotatingFile may be treated like a normal file-like object. Under the hood, however, it will create new sub-files and delete old ones.

The stream redirection of v4 (redirect_streams, start_log_fd_interception) is not available: the job supervisor copies a job's output into its log.

RotatingFile( file_path: pathlib.Path, num_files_to_keep: int | None = None, max_file_size: int | None = None, redirect_streams: bool = False, write_timestamps: bool = False, timestamp_format: str | None = None, write_callback=None) 
29    def __init__(
30        self,
31        file_path: pathlib.Path,
32        num_files_to_keep: Optional[int] = None,
33        max_file_size: Optional[int] = None,
34        redirect_streams: bool = False,
35        write_timestamps: bool = False,
36        timestamp_format: Optional[str] = None,
37        write_callback=None,
38    ):
39        """
40        Create a file-like object which manages other files.
41
42        Parameters
43        ----------
44        file_path: pathlib.Path
45            The base path (`logs/<name>.log`); subfiles are `<name>.log.<n>`.
46
47        num_files_to_keep: int, default None
48            How many sub-files to keep at any given time.
49            Defaults to the configured value (5).
50
51        max_file_size: int, default None
52            How large in bytes each sub-file can grow before another file is created.
53            Defaults to the configured value (100_000).
54
55        write_timestamps: bool, default False
56            If `True`, prepend the current UTC timestamp to each line of the file.
57
58        timestamp_format: str, default None
59            If `write_timestamps` is `True`, use this format for the timestamps.
60            Defaults to `'%Y-%m-%d %H:%M'`.
61        """
62        from meerschaum.config import get_config
63        from meerschaum.jobs import _native
64        if redirect_streams or write_callback is not None:
65            raise NotImplementedError(
66                "`redirect_streams` and `write_callback` are not supported in Meerschaum v5 "
67                "(see the migration guide, 'Removed in 5.0')."
68            )
69        self.file_path = pathlib.Path(file_path)
70        if num_files_to_keep is None:
71            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
72        if max_file_size is None:
73            max_file_size = get_config('jobs', 'logs', 'max_file_size')
74        if timestamp_format is None:
75            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
76        if num_files_to_keep < 1:
77            raise ValueError("At least 1 file must be kept.")
78        if max_file_size < 100:
79            raise ValueError("Subfiles must contain at least 100 bytes.")
80        self.num_files_to_keep = num_files_to_keep
81        self.max_file_size = max_file_size
82        self.write_timestamps = write_timestamps
83        self.timestamp_format = timestamp_format
84        self._core = _native.core().RotatingFile(
85            str(self.file_path),
86            num_files_to_keep,
87            max_file_size,
88            write_timestamps,
89            timestamp_format,
90        )
91        self._closed = False
92        self._lines_buffer: List[str] = []

Create a file-like object which manages other files.

Parameters
  • file_path (pathlib.Path): The base path (logs/<name>.log); subfiles are <name>.log.<n>.
  • num_files_to_keep (int, default None): How many sub-files to keep at any given time. Defaults to the configured value (5).
  • max_file_size (int, default None): How large in bytes each sub-file can grow before another file is created. Defaults to the configured value (100_000).
  • write_timestamps (bool, default False): If True, prepend the current UTC timestamp to each line of the file.
  • timestamp_format (str, default None): If write_timestamps is True, use this format for the timestamps. Defaults to '%Y-%m-%d %H:%M'.
SEEK_BACK_ATTEMPTS: int = 5
file_path
num_files_to_keep
max_file_size
write_timestamps
timestamp_format
closed: bool 
94    @property
95    def closed(self) -> bool:
96        return self._closed
def readable(self) -> bool: 
98    def readable(self) -> bool:
99        return True

Return whether object was opened for reading.

If False, read() will raise OSError.

def writable(self) -> bool: 
101    def writable(self) -> bool:
102        return True

Return whether object was opened for writing.

If False, write() will raise OSError.

def seekable(self) -> bool: 
104    def seekable(self) -> bool:
105        return True

Return whether object supports random access.

If False, seek(), tell() and truncate() will raise OSError. This method may need to do a test seek().

def isatty(self) -> bool: 
107    def isatty(self) -> bool:
108        return False

Return whether this is an 'interactive' stream.

Return False if it can't be determined.

def fileno(self) -> int: 
110    def fileno(self) -> int:
111        raise io.UnsupportedOperation("A RotatingFile has no file descriptor.")

Return underlying file descriptor if one exists.

Raise OSError if the IO object does not use a file descriptor.

def read(self, *args, **kw) -> str: 
113    def read(self, *args, **kw) -> str:
114        """Return the text of every subfile."""
115        return self._core.read()

Return the text of every subfile.

def readlines(self, *args, **kw) -> List[str]: 
117    def readlines(self, *args, **kw) -> List[str]:
118        """Return the lines since the cursor."""
119        return self._core.readlines()

Return the lines since the cursor.

def readline(self, *args, **kw) -> str: 
121    def readline(self, *args, **kw) -> str:
122        """Return the next line since the cursor ('' at the end)."""
123        if not self._lines_buffer:
124            self._lines_buffer = self._core.readlines()
125        return self._lines_buffer.pop(0) if self._lines_buffer else ''

Return the next line since the cursor ('' at the end).

def write(self, data: str) -> int: 
127    def write(self, data: str) -> int:
128        """Write text, rotating subfiles as they fill."""
129        return self._core.write(data)

Write text, rotating subfiles as they fill.

def writelines(self, lines) -> None: 
131    def writelines(self, lines) -> None:
132        for line in lines:
133            self.write(line)

Write a list of lines to stream.

Line separators are not added, so it is usual for each of the lines provided to have a line separator at the end.

def seek(self, position: int, *args, **kw) -> int: 
135    def seek(self, position: int, *args, **kw) -> int:
136        """Move the read cursor in the latest subfile."""
137        self._lines_buffer = []
138        self._core.seek(position)
139        return position

Move the read cursor in the latest subfile.

def touch(self) -> None: 
141    def touch(self) -> None:
142        """Create the latest subfile."""
143        self._core.touch()

Create the latest subfile.

def refresh_cursor(self) -> None: 
145    def refresh_cursor(self) -> None:
146        self._core.refresh_cursor()
def flush(self) -> None: 
148    def flush(self) -> None:
149        pass

Flush write buffers, if applicable.

This is not implemented for read-only and non-blocking streams.

def close(self) -> None: 
151    def close(self) -> None:
152        """Close the open subfile handles."""
153        self._core.close()
154        self._closed = True

Close the open subfile handles.

def delete(self, unused_only: bool = False) -> None: 
156    def delete(self, unused_only: bool = False) -> None:
157        """Delete the subfiles (`unused_only`: only those beyond the files to keep)."""
158        self._core.delete(unused_only)

Delete the subfiles (unused_only: only those beyond the files to keep).

def get_latest_subfile_index(self) -> int: 
160    def get_latest_subfile_index(self) -> int:
161        return self._core.get_latest_subfile_index()
def get_latest_subfile_path(self) -> pathlib.Path: 
163    def get_latest_subfile_path(self) -> pathlib.Path:
164        return pathlib.Path(self._core.get_latest_subfile_path())
def get_existing_subfile_indices(self) -> List[int]: 
166    def get_existing_subfile_indices(self) -> List[int]:
167        return self._core.get_existing_subfile_indices()
def get_existing_subfile_paths(self) -> List[pathlib.Path]: 
169    def get_existing_subfile_paths(self) -> List[pathlib.Path]:
170        return [pathlib.Path(p) for p in self._core.get_existing_subfile_paths()]
def get_subfile_path_from_index(self, index: int) -> pathlib.Path: 
172    def get_subfile_path_from_index(self, index: int) -> pathlib.Path:
173        return pathlib.Path(self._core.get_subfile_path_from_index(index))
def get_subfile_name_from_index(self, index: int) -> str: 
175    def get_subfile_name_from_index(self, index: int) -> str:
176        return self._core.get_subfile_name_from_index(index)
def get_index_from_subfile_name(self, subfile_name: str) -> int: 
178    def get_index_from_subfile_name(self, subfile_name: str) -> int:
179        return self._core.get_index_from_subfile_name(subfile_name)
def get_timestamp_prefix_str(self) -> str: 
181    def get_timestamp_prefix_str(self) -> str:
182        return self._core.get_timestamp_prefix_str()