Skip to content

yggdrasil.environ.shutdown

shutdown

yggdrasil.environ.shutdown

Shutdown callback registry for process exit and termination signals.

Callbacks are run on: - normal interpreter exit via atexit - termination-related signals such as SIGINT / SIGTERM / SIGHUP / SIGBREAK when available on the current platform and installable from the current thread

Design notes

  • Signal handlers do NOT raise synthetic SystemExit into arbitrary stack frames. Instead they run callbacks, restore the previous handler, and re-raise the signal to the process so the OS / previous handler decides what happens next. This avoids injecting exceptions into finally blocks and C extensions.
  • run() is both run-once and re-entrancy safe: a signal arriving while atexit callbacks are executing will NOT start a parallel run, and will NOT truncate the in-flight run.
  • Bound methods are tracked via weakref.WeakMethod so registering an instance method does not extend the instance's lifetime, and does not suffer from id() reuse after GC.
  • atexit is registered eagerly at module import so ordering relative to other libraries is predictable (LIFO at import time).

Typical usage

from yggdrasil.environ import shutdown as yg_shutdown

yg_shutdown.register(cleanup)
yg_shutdown.unregister(cleanup)

ShutdownRegistry 

ShutdownRegistry(
    *,
    handled_signals: Iterable[int] | None = None,
    raise_on_callback_error: bool = False
)

Registry for cleanup callbacks triggered by atexit and termination signals.

install 

install() -> None

Install the atexit hook and signal handlers.

Safe to call multiple times. If called outside the main thread, atexit is still registered but signal handlers are skipped.

uninstall 

uninstall() -> None

Restore previous signal handlers.

atexit registration is intentionally left in place; run() is idempotent so leaving the runner registered is harmless.

register 

register(
    callback: F,
    *,
    priority: int = 0,
    pass_reason: bool = False,
    install: bool = True
) -> F

register(
    callback: None = None,
    *,
    priority: int = 0,
    pass_reason: bool = False,
    install: bool = True
) -> Callable[[F], F]

register(
    callback: Callback | None = None,
    *,
    priority: int = 0,
    pass_reason: bool = False,
    install: bool = True
)

Register a callback.

If pass_reason=True, the callback will receive one positional argument: reason: str

Bound methods are held via weakref so the registry does not extend the instance's lifetime.

unregister 

unregister(callback: Callback) -> bool

Unregister a callback. Returns True if removed.

clear 

clear() -> None

Remove all registered callbacks.

callbacks 

callbacks() -> tuple[Callback, ...]

Return currently live callbacks in execution order.

Dead weak references are skipped (but not pruned here; pruning happens in run() or via the WeakMethod finalizer).

run 

run(reason: str = 'manual') -> None

Run registered callbacks once.

Re-entrancy safe: if a signal arrives while this is already running, the nested call is a no-op and does NOT truncate the in-flight run.

Subsequent top-level calls are also no-ops until reset_run_state() is called.

reset_run_state 

reset_run_state() -> None

Allow callbacks to run again. Mainly useful for tests.

register 

register(
    callback: Callback | None = None,
    *,
    priority: int = 0,
    pass_reason: bool = False
)

Convenience alias for the shared registry register().

unregister 

unregister(callback: Callback) -> bool

Convenience alias for the shared registry unregister().