meerschaum.utils.daemon.RotatingFile

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