#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Neutral post-save / post-load hook registry for scitex-io.
Public extension API. Observer packages (scitex-clew, future
notification or audit observers) register callbacks here on their own
import; scitex-io itself never names any observer.
Per SOC.md R6: producer packages don't know observer packages. The
arrow runs observer → producer. scitex-io owns this registry; observers
self-register.
Usage from an observer
----------------------
>>> from scitex_io import register_post_save_hook
>>> def on_save(path, obj, kwargs):
... # never raise; hooks fail silently
... ...
>>> register_post_save_hook(on_save)
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Callable, List
logger = logging.getLogger(__name__)
PostSaveHook = Callable[[Path, Any, dict], None]
"""Signature: ``(saved_path, obj, kwargs) -> None``."""
PostLoadHook = Callable[[Path, Any], None]
"""Signature: ``(loaded_path, result) -> None``."""
_post_save_hooks: List[PostSaveHook] = []
_post_load_hooks: List[PostLoadHook] = []
[docs]
def register_post_save_hook(fn: PostSaveHook) -> None:
"""Register a function to run after every successful ``scitex_io.save``.
Hooks fire in registration order. They MUST NOT raise — exceptions
are swallowed with ``logger.debug``. A misbehaving observer must
never break the host's I/O.
"""
_post_save_hooks.append(fn)
[docs]
def register_post_load_hook(fn: PostLoadHook) -> None:
"""Register a function to run after every successful ``scitex_io.load``."""
_post_load_hooks.append(fn)
def fire_post_save(path: Path, obj: Any, kwargs: dict) -> None:
"""Internal: invoke registered post-save hooks. Never raises."""
# Diagnosability (auto-provenance incident, 2026-07-04): log the FIRING
# instance's fingerprint so a single DEBUG smoke pinpoints a dead hook.
# `count == 0` here while an observer logged a successful registration
# ⇒ module-identity split (it registered on a DIFFERENT scitex_io
# instance's list than the one firing). Compare `hooks_id` / `module`
# against the observer's own registration-side DEBUG log.
if logger.isEnabledFor(logging.DEBUG):
logger.debug(
"fire_post_save: %d hook(s) for %s; _post_save_hooks id=%d module=%s",
len(_post_save_hooks),
path,
id(_post_save_hooks),
__file__,
)
for fn in _post_save_hooks:
try:
fn(path, obj, kwargs)
except Exception as e: # noqa: BLE001
logger.debug("post-save hook %r failed for %s: %s", fn, path, e)
def fire_post_load(path: Path, result: Any) -> None:
"""Internal: invoke registered post-load hooks. Never raises."""
for fn in _post_load_hooks:
try:
fn(path, result)
except Exception as e: # noqa: BLE001
logger.debug("post-load hook %r failed for %s: %s", fn, path, e)
__all__ = [
"PostSaveHook",
"PostLoadHook",
"register_post_save_hook",
"register_post_load_hook",
"fire_post_save",
"fire_post_load",
]