Skip to content

reader

Audit trail reader for querying and filtering historical audit events.

AuditTrailReader

Stream audit events from JSONL segments with optional filtering.

Source code in src/horde_model_reference/audit/reader.py 
 19
 20
 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
class AuditTrailReader:
    """Stream audit events from JSONL segments with optional filtering."""

    def __init__(self, *, root_path: Path) -> None:
        """Initialize the reader with the audit root directory."""
        self._root_path = Path(root_path)

    def iter_events(
        self,
        *,
        domains: Collection[CanonicalFormat] | None = None,
        categories: Collection[str] | None = None,
        model_names: Collection[str] | None = None,
        min_event_id: int | None = None,
        max_event_id: int | None = None,
        min_timestamp: int | None = None,
        max_timestamp: int | None = None,
    ) -> Iterator[AuditEvent]:
        """Yield AuditEvent objects matching the provided filters."""
        if not self._root_path.exists():
            return

        domain_filter = set(domains) if domains else None
        category_filter = set(categories) if categories else None
        model_filter = set(model_names) if model_names else None

        for domain_path in _iter_dirs(self._root_path):
            try:
                domain = CanonicalFormat(domain_path.name)
            except ValueError:
                logger.debug(f"Skipping unknown audit domain directory: {domain_path}")
                continue

            if domain_filter and domain not in domain_filter:
                continue

            for category_path in _iter_dirs(domain_path):
                category = category_path.name
                if category_filter and category not in category_filter:
                    continue

                for segment_path in sorted(category_path.glob("audit-*.jsonl")):
                    yield from self._iter_segment(
                        segment_path,
                        domain=domain,
                        category=category,
                        model_filter=model_filter,
                        min_event_id=min_event_id,
                        max_event_id=max_event_id,
                        min_timestamp=min_timestamp,
                        max_timestamp=max_timestamp,
                    )

    def _iter_segment(
        self,
        segment_path: Path,
        *,
        domain: CanonicalFormat,
        category: str,
        model_filter: set[str] | None,
        min_event_id: int | None,
        max_event_id: int | None,
        min_timestamp: int | None,
        max_timestamp: int | None,
    ) -> Iterator[AuditEvent]:
        try:
            with segment_path.open(encoding="utf-8") as handle:
                for line in handle:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        event = AuditEvent.model_validate_json(line)
                    except (ValidationError, ValueError) as exc:
                        logger.warning(f"Skipping malformed audit event in {segment_path}: {exc}")
                        continue

                    if event.domain != domain or event.category != category:
                        continue

                    if model_filter and event.model_name not in model_filter:
                        continue

                    if min_event_id is not None and event.event_id < min_event_id:
                        continue

                    if max_event_id is not None and event.event_id > max_event_id:
                        continue

                    if min_timestamp is not None and event.timestamp < min_timestamp:
                        continue

                    if max_timestamp is not None and event.timestamp > max_timestamp:
                        continue

                    yield event
        except FileNotFoundError:
            logger.warning(f"Audit segment disappeared during iteration: {segment_path}")
        except OSError as exc:
            logger.warning(f"Unable to read audit segment {segment_path}: {exc}")

_root_path instance-attribute 

_root_path = Path(root_path)

__init__ 

__init__(*, root_path: Path) -> None

Initialize the reader with the audit root directory.

Source code in src/horde_model_reference/audit/reader.py 
22
23
24
def __init__(self, *, root_path: Path) -> None:
    """Initialize the reader with the audit root directory."""
    self._root_path = Path(root_path)

iter_events 

iter_events(
    *,
    domains: Collection[CanonicalFormat] | None = None,
    categories: Collection[str] | None = None,
    model_names: Collection[str] | None = None,
    min_event_id: int | None = None,
    max_event_id: int | None = None,
    min_timestamp: int | None = None,
    max_timestamp: int | None = None,
) -> Iterator[AuditEvent]

Yield AuditEvent objects matching the provided filters.

Source code in src/horde_model_reference/audit/reader.py 
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
def iter_events(
    self,
    *,
    domains: Collection[CanonicalFormat] | None = None,
    categories: Collection[str] | None = None,
    model_names: Collection[str] | None = None,
    min_event_id: int | None = None,
    max_event_id: int | None = None,
    min_timestamp: int | None = None,
    max_timestamp: int | None = None,
) -> Iterator[AuditEvent]:
    """Yield AuditEvent objects matching the provided filters."""
    if not self._root_path.exists():
        return

    domain_filter = set(domains) if domains else None
    category_filter = set(categories) if categories else None
    model_filter = set(model_names) if model_names else None

    for domain_path in _iter_dirs(self._root_path):
        try:
            domain = CanonicalFormat(domain_path.name)
        except ValueError:
            logger.debug(f"Skipping unknown audit domain directory: {domain_path}")
            continue

        if domain_filter and domain not in domain_filter:
            continue

        for category_path in _iter_dirs(domain_path):
            category = category_path.name
            if category_filter and category not in category_filter:
                continue

            for segment_path in sorted(category_path.glob("audit-*.jsonl")):
                yield from self._iter_segment(
                    segment_path,
                    domain=domain,
                    category=category,
                    model_filter=model_filter,
                    min_event_id=min_event_id,
                    max_event_id=max_event_id,
                    min_timestamp=min_timestamp,
                    max_timestamp=max_timestamp,
                )

_iter_segment 

_iter_segment(
    segment_path: Path,
    *,
    domain: CanonicalFormat,
    category: str,
    model_filter: set[str] | None,
    min_event_id: int | None,
    max_event_id: int | None,
    min_timestamp: int | None,
    max_timestamp: int | None,
) -> Iterator[AuditEvent]
Source code in src/horde_model_reference/audit/reader.py 
 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
def _iter_segment(
    self,
    segment_path: Path,
    *,
    domain: CanonicalFormat,
    category: str,
    model_filter: set[str] | None,
    min_event_id: int | None,
    max_event_id: int | None,
    min_timestamp: int | None,
    max_timestamp: int | None,
) -> Iterator[AuditEvent]:
    try:
        with segment_path.open(encoding="utf-8") as handle:
            for line in handle:
                line = line.strip()
                if not line:
                    continue
                try:
                    event = AuditEvent.model_validate_json(line)
                except (ValidationError, ValueError) as exc:
                    logger.warning(f"Skipping malformed audit event in {segment_path}: {exc}")
                    continue

                if event.domain != domain or event.category != category:
                    continue

                if model_filter and event.model_name not in model_filter:
                    continue

                if min_event_id is not None and event.event_id < min_event_id:
                    continue

                if max_event_id is not None and event.event_id > max_event_id:
                    continue

                if min_timestamp is not None and event.timestamp < min_timestamp:
                    continue

                if max_timestamp is not None and event.timestamp > max_timestamp:
                    continue

                yield event
    except FileNotFoundError:
        logger.warning(f"Audit segment disappeared during iteration: {segment_path}")
    except OSError as exc:
        logger.warning(f"Unable to read audit segment {segment_path}: {exc}")

_iter_dirs 

_iter_dirs(path: Path) -> list[Path]
Source code in src/horde_model_reference/audit/reader.py 
15
16
def _iter_dirs(path: Path) -> list[Path]:
    return [child for child in sorted(path.iterdir()) if child.is_dir()]