#!/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."""
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",
]