Skip to content

sgnts.sinks.retention

Mixin that adds file retention policies to sinks that write files.

FileRetentionMixin dataclass

Bases: HasLogger


              flowchart TD
              sgnts.sinks.retention.FileRetentionMixin[FileRetentionMixin]
              sgnts.sinks.retention.HasLogger[HasLogger]

                              sgnts.sinks.retention.HasLogger --> sgnts.sinks.retention.FileRetentionMixin
                


              click sgnts.sinks.retention.FileRetentionMixin href "" "sgnts.sinks.retention.FileRetentionMixin"
              click sgnts.sinks.retention.HasLogger href "" "sgnts.sinks.retention.HasLogger"

Mixin that adds file retention policies to any sink that writes files.

Provides count-based and time-based retention, which can be used independently or combined. After each file write, the sink calls :meth:track_file to register the path; cleanup runs automatically.

Call :meth:clean_up_directory during startup (e.g. in configure) to adopt files left by a previous run whose tracking state was lost, so the retention policies apply to them as well. Pass it a pattern that matches only this sink's own files if the output directory is shared with other writers.

Parameters:

Name Type Description Default
max_files int | None

Keep only the N most recent files. Older files are deleted when the count is exceeded. Disabled when None.

 None
retention_time float | None

Retention time in seconds. Files whose mtime is older than this are deleted. Can be combined with max_files. Disabled when None.

 None

Attributes:

Name Type Description
files_reaped int

Running count of files deleted by the retention policies, including any deleted while adopting leftovers at startup. Sinks can snapshot this to report reap counts in their own periodic summaries.
Source code in src/sgnts/sinks/retention.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@dataclass(kw_only=True)
class FileRetentionMixin(HasLogger):
    """Mixin that adds file retention policies to any sink that writes files.

    Provides count-based and time-based retention, which can be used
    independently or combined. After each file write, the sink calls
    :meth:`track_file` to register the path; cleanup runs automatically.

    Call :meth:`clean_up_directory` during startup (e.g. in ``configure``)
    to adopt files left by a previous run whose tracking state was lost,
    so the retention policies apply to them as well.  Pass it a pattern
    that matches only this sink's own files if the output directory is
    shared with other writers.

    Args:
        max_files:
            Keep only the N most recent files. Older files are deleted
            when the count is exceeded. Disabled when ``None``.
        retention_time:
            Retention time in seconds. Files whose mtime is older than
            this are deleted. Can be combined with *max_files*. Disabled
            when ``None``.

    Attributes:
        files_reaped:
            Running count of files deleted by the retention policies,
            including any deleted while adopting leftovers at startup.
            Sinks can snapshot this to report reap counts in their own
            periodic summaries.
    """

    max_files: int | None = None
    retention_time: float | None = None

    files_reaped: int = field(default=0, init=False, repr=False)

    _file_cache: deque[str] = field(default_factory=deque, init=False, repr=False)
    _reap_logged: bool = field(default=False, init=False, repr=False)

    def track_file(self, path: str | Path) -> None:
        """Register a written file and run cleanup if policies are set."""
        self._file_cache.append(str(path))
        if self.max_files is not None or self.retention_time is not None:
            deleted = self.clean_up_files()
            # The first reap is a state transition worth an INFO line;
            # after that, steady-state rolling cleanup stays at DEBUG so
            # a healthy run is quiet
            if deleted > 0 and not self._reap_logged:
                self._reap_logged = True
                self.logger.info(
                    "Retention reaping started: deleted %d file(s)", deleted
                )
            elif deleted > 0:
                self.logger.debug("Cleaned up %d old files", deleted)

    def clean_up_files(self) -> int:
        """Remove tracked files that exceed retention policies.

        Returns:
            Number of files deleted.
        """
        deleted = 0
        now = time.time()

        if self.retention_time:
            while self._file_cache:
                path = Path(self._file_cache[0])
                try:
                    is_old = (now - path.stat().st_mtime) > self.retention_time
                except FileNotFoundError:
                    self._file_cache.popleft()
                    deleted += 1
                    continue
                if not is_old:
                    break
                try:
                    path.unlink()
                    self.logger.debug("Removed old file: %s", path)
                    deleted += 1
                except FileNotFoundError:
                    deleted += 1
                except OSError:
                    self.logger.exception("Error deleting file %s", path)
                self._file_cache.popleft()

        if self.max_files and self.max_files > 0:
            while len(self._file_cache) > self.max_files:
                path = Path(self._file_cache.popleft())
                try:
                    path.unlink()
                    deleted += 1
                    self.logger.debug("Removed old file: %s", path)
                except FileNotFoundError:
                    pass
                except OSError:
                    self.logger.exception("Error deleting file %s", path)

        self.files_reaped += deleted
        return deleted

    def clean_up_directory(
        self,
        directory: str | Path,
        pattern: str,
        *,
        recursive: bool = False,
    ) -> None:
        """Adopt files in *directory* matching *pattern* and apply retention policies.

        This is useful after a restart when the in-memory file cache has
        been lost.  Matching files are added to the file cache in
        modification-time order and the retention policies applied, so
        leftovers from a previous run behave exactly like files this run
        wrote itself: files already past *retention_time* are deleted
        immediately, younger ones age out as the run proceeds, and
        *max_files* counts them toward its limit.

        When several processes write to one directory, give a *pattern*
        specific enough to match only this sink's own files, so that a
        restarting sink does not sweep away another's output.

        Args:
            directory:
                Directory to scan.
            pattern:
                Glob matched against file names, e.g. ``"H-SGN-*-*.gwf"``.
                A value with no glob characters is taken as a plain
                suffix (e.g. ``".gwf"``, ``".xml.gz"``), matching any
                file that ends with it, so compound extensions work.
            recursive:
                Whether to descend into subdirectories.  Default False.
        """
        if not self.retention_time and not self.max_files:
            return
        directory = Path(directory)
        if not directory.exists():
            return
        if not any(char in pattern for char in "*?["):
            pattern = f"*{pattern}"
        walk = directory.rglob if recursive else directory.glob
        found = []
        for entry in walk(pattern):
            try:
                if entry.is_file():
                    found.append((entry.stat().st_mtime, str(entry)))
            except OSError:
                continue
        found.sort()
        self._file_cache.extend(path for _, path in found)
        deleted = self.clean_up_files()
        # Always log the startup scan, even when nothing matched: the
        # line confirms which directory was scanned and that retention
        # is active
        self.logger.info(
            "Adopted %d leftover file(s) from %s (deleted %d, tracking %d)",
            len(found),
            directory,
            deleted,
            len(self._file_cache),
        )

clean_up_directory(directory, pattern, *, recursive=False)

Adopt files in directory matching pattern and apply retention policies.

This is useful after a restart when the in-memory file cache has been lost. Matching files are added to the file cache in modification-time order and the retention policies applied, so leftovers from a previous run behave exactly like files this run wrote itself: files already past retention_time are deleted immediately, younger ones age out as the run proceeds, and max_files counts them toward its limit.

When several processes write to one directory, give a pattern specific enough to match only this sink's own files, so that a restarting sink does not sweep away another's output.

Parameters:

Name Type Description Default
directory str | Path

Directory to scan.

 required
pattern str

Glob matched against file names, e.g. "H-SGN-*-*.gwf". A value with no glob characters is taken as a plain suffix (e.g. ".gwf", ".xml.gz"), matching any file that ends with it, so compound extensions work.

 required
recursive bool

Whether to descend into subdirectories. Default False.

 False
Source code in src/sgnts/sinks/retention.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def clean_up_directory(
    self,
    directory: str | Path,
    pattern: str,
    *,
    recursive: bool = False,
) -> None:
    """Adopt files in *directory* matching *pattern* and apply retention policies.

    This is useful after a restart when the in-memory file cache has
    been lost.  Matching files are added to the file cache in
    modification-time order and the retention policies applied, so
    leftovers from a previous run behave exactly like files this run
    wrote itself: files already past *retention_time* are deleted
    immediately, younger ones age out as the run proceeds, and
    *max_files* counts them toward its limit.

    When several processes write to one directory, give a *pattern*
    specific enough to match only this sink's own files, so that a
    restarting sink does not sweep away another's output.

    Args:
        directory:
            Directory to scan.
        pattern:
            Glob matched against file names, e.g. ``"H-SGN-*-*.gwf"``.
            A value with no glob characters is taken as a plain
            suffix (e.g. ``".gwf"``, ``".xml.gz"``), matching any
            file that ends with it, so compound extensions work.
        recursive:
            Whether to descend into subdirectories.  Default False.
    """
    if not self.retention_time and not self.max_files:
        return
    directory = Path(directory)
    if not directory.exists():
        return
    if not any(char in pattern for char in "*?["):
        pattern = f"*{pattern}"
    walk = directory.rglob if recursive else directory.glob
    found = []
    for entry in walk(pattern):
        try:
            if entry.is_file():
                found.append((entry.stat().st_mtime, str(entry)))
        except OSError:
            continue
    found.sort()
    self._file_cache.extend(path for _, path in found)
    deleted = self.clean_up_files()
    # Always log the startup scan, even when nothing matched: the
    # line confirms which directory was scanned and that retention
    # is active
    self.logger.info(
        "Adopted %d leftover file(s) from %s (deleted %d, tracking %d)",
        len(found),
        directory,
        deleted,
        len(self._file_cache),
    )

clean_up_files()

Remove tracked files that exceed retention policies.

Returns:

Type Description
int

Number of files deleted.
Source code in src/sgnts/sinks/retention.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def clean_up_files(self) -> int:
    """Remove tracked files that exceed retention policies.

    Returns:
        Number of files deleted.
    """
    deleted = 0
    now = time.time()

    if self.retention_time:
        while self._file_cache:
            path = Path(self._file_cache[0])
            try:
                is_old = (now - path.stat().st_mtime) > self.retention_time
            except FileNotFoundError:
                self._file_cache.popleft()
                deleted += 1
                continue
            if not is_old:
                break
            try:
                path.unlink()
                self.logger.debug("Removed old file: %s", path)
                deleted += 1
            except FileNotFoundError:
                deleted += 1
            except OSError:
                self.logger.exception("Error deleting file %s", path)
            self._file_cache.popleft()

    if self.max_files and self.max_files > 0:
        while len(self._file_cache) > self.max_files:
            path = Path(self._file_cache.popleft())
            try:
                path.unlink()
                deleted += 1
                self.logger.debug("Removed old file: %s", path)
            except FileNotFoundError:
                pass
            except OSError:
                self.logger.exception("Error deleting file %s", path)

    self.files_reaped += deleted
    return deleted

track_file(path)

Register a written file and run cleanup if policies are set.

Source code in src/sgnts/sinks/retention.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def track_file(self, path: str | Path) -> None:
    """Register a written file and run cleanup if policies are set."""
    self._file_cache.append(str(path))
    if self.max_files is not None or self.retention_time is not None:
        deleted = self.clean_up_files()
        # The first reap is a state transition worth an INFO line;
        # after that, steady-state rolling cleanup stays at DEBUG so
        # a healthy run is quiet
        if deleted > 0 and not self._reap_logged:
            self._reap_logged = True
            self.logger.info(
                "Retention reaping started: deleted %d file(s)", deleted
            )
        elif deleted > 0:
            self.logger.debug("Cleaned up %d old files", deleted)