HookManager

class brainstate.HookManager(config=None)

Manager for hooks on a single State instance.

This class handles registration, unregistration, and execution of hooks for all hook types (read, write_before, write_after, restore).

Features:

• Priority-based execution ordering

• Thread-safe hook registration and execution

• Hook list caching for performance

• Sequential chaining for write_before hooks

• Configurable error handling

Thread Safety:

All operations are protected by a reentrant lock (RLock), allowing hooks to safely trigger other state operations.

Examples

>>> from brainstate import HookManager
>>> manager = HookManager()
>>> handle = manager.register_hook('read', lambda ctx: print(ctx.value), priority=10)
>>> manager.has_hooks('read')
True
>>> handle.disable()
>>> manager.clear_hooks()

clear_hooks(hook_type=None)

Clear all hooks, optionally filtered by type.

Parameters:

hook_type (str | None) – Optional hook type to clear (if None, clears all)

Return type:

None

execute_init_hooks(value, state_ref, init_metadata=None)

Execute all init hooks.

Parameters:

• value (Any) – The initial value of the state

• state_ref (ReferenceType) – Weak reference to the State instance

• init_metadata (Dict[str, Any] | None) – Optional dictionary of initialization metadata

Return type:

None

execute_read_hooks(value, state_ref)

Execute all read hooks.

Parameters:

• value (Any) – The value being read

• state_ref (ReferenceType) – Weak reference to the State instance

Return type:

None

execute_restore_hooks(new_value, old_value, state_ref)

Execute all restore hooks.

Parameters:

• new_value (Any) – The restored value

• old_value (Any) – The previous value before restoration

• state_ref (ReferenceType) – Weak reference to the State instance

Return type:

None

execute_write_after_hooks(new_value, old_value, state_ref)

Execute all write_after hooks.

Parameters:

• new_value (Any) – The new value that was written

• old_value (Any) – The previous value

• state_ref (ReferenceType) – Weak reference to the State instance

Return type:

None

execute_write_before_hooks(new_value, old_value, state_ref)

Execute all write_before hooks with sequential chaining.

Each hook can transform the value, and the next hook receives the transformed output.

Parameters:

• new_value (Any) – The new value being written

• old_value (Any) – The previous value

• state_ref (ReferenceType) – Weak reference to the State instance

Return type:

Any

Raises:

HookCancellationError – If a hook cancels the operation

get_hooks(hook_type=None)

Get all registered hooks, optionally filtered by type.

Parameters:

hook_type (str | None) – Optional hook type to filter by

Return type:

List[Hook]

has_hooks(hook_type=None)

Check if any hooks are registered.

Parameters:

hook_type (str | None) – Optional hook type to check (if None, checks all)

Return type:

bool

Raises:

ValueError – If hook_type is not a recognised hook type.

register_hook(hook_type, callback, priority=0, name=None, enabled=True)

Register a new hook.

Parameters:

• hook_type (Literal['read', 'write_before', 'write_after', 'restore', 'init']) – Type of hook to register

• callback (Callable[[HookContext], Any]) – Callable that receives a HookContext

• priority (int) – Execution priority (higher = earlier, default 0)

• name (str | None) – Optional name for the hook

• enabled (bool) – Whether the hook is enabled initially (default True)

Return type:

HookHandle

Raises:

HookRegistrationError – If hook_type is invalid

Examples

>>> def my_hook(ctx):
...     print(f"Hook called: {ctx.operation}")
>>> handle = manager.register_hook('read', my_hook, priority=10, name='my_reader')

unregister_hook(handle)

Unregister a hook using its handle.

Parameters:

handle (HookHandle) – The HookHandle to unregister

Return type:

bool