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 sweep stale files left by a previous run whose tracking state was lost.

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
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
@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 sweep stale files left by a previous run whose tracking state was
    lost.

    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``.
    """

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

    _file_cache: deque[str] = field(default_factory=deque, 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()
            if deleted > 0:
                self.logger.info("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)

        return deleted

    def clean_up_directory(self, directory: str | Path, suffix: str) -> None:
        """Remove files in *directory* matching *suffix* that exceed *retention_time*.

        This is useful after a restart when the in-memory file cache has
        been lost.  Only time-based retention applies here — count-based
        cleanup requires creation-order knowledge that a directory scan
        cannot reliably provide.

        Args:
            directory:
                Directory to scan.
            suffix:
                File suffix to match (e.g. ``".gwf"``, ``".xml.gz"``).
                Matched with ``str.endswith`` so compound extensions work.
        """
        if not self.retention_time:
            return
        directory = Path(directory)
        if not directory.exists():
            return
        now = time.time()
        for entry in directory.iterdir():
            if entry.name.endswith(suffix):
                try:
                    if (now - entry.stat().st_mtime) > self.retention_time:
                        entry.unlink(missing_ok=True)
                        self.logger.debug("Removed old file: %s", entry)
                except OSError:
                    self.logger.exception("Error deleting file %s", entry)

clean_up_directory(directory, suffix)

Remove files in directory matching suffix that exceed retention_time.

This is useful after a restart when the in-memory file cache has been lost. Only time-based retention applies here — count-based cleanup requires creation-order knowledge that a directory scan cannot reliably provide.

Parameters:

Name Type Description Default
directory str | Path

Directory to scan.

 required
suffix str

File suffix to match (e.g. ".gwf", ".xml.gz"). Matched with str.endswith so compound extensions work.

 required
Source code in src/sgnts/sinks/retention.py 
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
def clean_up_directory(self, directory: str | Path, suffix: str) -> None:
    """Remove files in *directory* matching *suffix* that exceed *retention_time*.

    This is useful after a restart when the in-memory file cache has
    been lost.  Only time-based retention applies here — count-based
    cleanup requires creation-order knowledge that a directory scan
    cannot reliably provide.

    Args:
        directory:
            Directory to scan.
        suffix:
            File suffix to match (e.g. ``".gwf"``, ``".xml.gz"``).
            Matched with ``str.endswith`` so compound extensions work.
    """
    if not self.retention_time:
        return
    directory = Path(directory)
    if not directory.exists():
        return
    now = time.time()
    for entry in directory.iterdir():
        if entry.name.endswith(suffix):
            try:
                if (now - entry.stat().st_mtime) > self.retention_time:
                    entry.unlink(missing_ok=True)
                    self.logger.debug("Removed old file: %s", entry)
            except OSError:
                self.logger.exception("Error deleting file %s", entry)

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 
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
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)

    return deleted

track_file(path)

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

Source code in src/sgnts/sinks/retention.py 
48
49
50
51
52
53
54
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()
        if deleted > 0:
            self.logger.info("Cleaned up %d old files", deleted)