Skip to content

fathom.FileSink

fathom.FileSink

Writes audit records as JSON Lines to a file (append mode).

Source code in src/fathom/audit.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class FileSink:
    """Writes audit records as JSON Lines to a file (append mode)."""

    def __init__(self, path: str | Path) -> None:
        """Create a file sink.

        Args:
            path: Path to the JSON Lines audit file (created if missing).
        """
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._path.touch(exist_ok=True)

    def write(self, record: AuditPayload) -> None:
        if isinstance(record, AuditRecord):
            payload = record.model_dump_json()
        else:
            # Non-evaluation events (hot reload) arrive as plain mappings.
            # ``default=str`` keeps a stray datetime or Path from making a
            # reload unwritable.
            payload = json.dumps(dict(record), default=str)
        with open(self._path, "a", encoding="utf-8") as f:
            f.write(payload + "\n")

__init__(path)

Create a file sink.

Parameters:

Name Type Description Default
path str | Path

Path to the JSON Lines audit file (created if missing).

 required
Source code in src/fathom/audit.py 
33
34
35
36
37
38
39
40
41
def __init__(self, path: str | Path) -> None:
    """Create a file sink.

    Args:
        path: Path to the JSON Lines audit file (created if missing).
    """
    self._path = Path(path)
    self._path.parent.mkdir(parents=True, exist_ok=True)
    self._path.touch(exist_ok=True)