Audit

Structured catalog audit events and JSON Lines storage.

class ogcat.audit.AuditEvent(operation_id, level, event_type, message, event_id=<factory>, timestamp_utc=<factory>, user_id=None, catalog_id=None, catalog_path=None, record_id=None, locator=None, details=<factory>, exception_type=None, exception_message=None, traceback=None)

Bases: object

One structured audit event emitted by a catalog operation.

Parameters:

• event_id (str) – Unique event identifier.

• operation_id (str) – Operation identifier shared with the active transaction.

• timestamp_utc (str) – UTC ISO 8601 timestamp.

• level (str) – Event severity, such as "info", "warning", or "error".

• event_type (str) – Stable lifecycle event type.

• user_id (str | None) – User associated with the operation.

• catalog_id (str | None) – Catalog name or other stable catalog identifier.

• catalog_path (str | None) – Filesystem path to the catalog root.

• record_id (str | None) – Catalog record id touched by the event, if known.

• locator (MetadataDict | None) – Artifact locator summary, if known.

• message (str) – Human-readable event summary.

• details (MetadataDict) – Redacted JSON-compatible event details.

• exception_type (str | None) – Exception class name for failures.

• exception_message (str | None) – Exception message for failures.

• traceback (str | None) – Optional truncated traceback text.

operation_id: str

level: str

event_type: str

message: str

event_id: str

timestamp_utc: str

user_id: str | None

catalog_id: str | None

catalog_path: str | None

record_id: str | None

locator: dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

details: dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]]

exception_type: str | None

exception_message: str | None

traceback: str | None

to_dict()

Return the event as a JSON-compatible dictionary.

Return type:

MetadataDict

classmethod from_dict(data)

Build an event from a dictionary read from JSON Lines.

Return type:

AuditEvent

classmethod from_exception(*, operation_id, event_type, message, exception, user_id=None, catalog_id=None, catalog_path=None, record_id=None, locator=None, details=None, traceback_limit=8)

Build an error event from an exception.

Return type:

AuditEvent

class ogcat.audit.AuditSink(*args, **kwargs)

Bases: Protocol

Sink for structured audit events.

emit(event)

Persist or forward one audit event.

Return type:

None

class ogcat.audit.JsonlAuditSink(catalog_root, relative_path=PosixPath('.ogcat/logs/events.jsonl'))

Bases: object

Append-only JSON Lines audit sink under a catalog root.

Parameters:

• catalog_root (Path) – Catalog root directory.

• relative_path (Path) – Log path relative to catalog_root.

catalog_root: Path

relative_path: Path

property path: Path

Absolute path to the JSON Lines audit log.

emit(event)

Append one event as a JSON line.

Return type:

None

read_events(*, user_id=None, operation_id=None, record_id=None, level=None, event_type=None, limit=None)

Read events from the sink with optional filters.

Return type:

list[AuditEvent]

ogcat.audit.add_operation_note(exception, operation_id)

Attach an operation-id note to an exception if one is not already present.

Return type:

None

ogcat.audit.exception_operation_id(exception)

Return an operation id attached to an exception note, if present.

Return type:

str | None

ogcat.audit.read_audit_events(path, *, user_id=None, operation_id=None, record_id=None, level=None, event_type=None, limit=None)

Read matching audit events from a JSON Lines file.

Corrupt JSON lines and malformed event dictionaries are skipped with a runtime warning so one bad line does not hide the rest of the audit log.

Return type:

list[AuditEvent]

ogcat.audit.redact_sensitive_values(value)

Return a metadata dictionary with sensitive-looking keys redacted.

Return type:

MetadataDict