duck.utils.threading.patch

Production-grade monkey-patch module for Python’s threading.Thread.

Enhancements:

  • Tracks parent thread object (not just ident).

  • Returns parent Thread object via get_parent_thread().

  • Uses strong weakref-based registry (no leaks).

  • Supports patching already-created Thread instances (before start()).

  • Preserves subclassed run() methods.

  • Hooks for pre-run and post-run execution.

  • Fully idempotent. Safe for large production systems.

  • Automatic cleanup after thread finishes.

Module Contents

Functions

_wrap_run

Returns a wrapped run() method that executes:

get_parent_thread

Returns the actual parent Thread object of a given thread.

patch_threading

Monkey-patches threading.Thread so that:

Data

_is_patched

_original_init

get_parent

thread_info

API

exception duck.utils.threading.patch.PatchNotApplied[source]

Bases: Exception

Raised if user tries to use get_parent but forgot to patch threading module.

Initialization

Initialize self. See help(type(self)) for accurate signature.

duck.utils.threading.patch._is_patched

False

duck.utils.threading.patch._original_init

None

duck.utils.threading.patch._wrap_run(self, original_run, pre_hook, post_hook)[source]

Returns a wrapped run() method that executes:

  • pre_hook()

  • original run()

  • post_hook()

  • registry cleanup

duck.utils.threading.patch.get_parent

None

duck.utils.threading.patch.get_parent_thread(thread_or_ident: Union[int, threading.Thread]) → Optional[threading.Thread][source]

Returns the actual parent Thread object of a given thread.

Parameters:

thread_or_ident – A thread object or its ident.

Returns:

Thread | None

Raises:

PatchNotApplied – if thread module wasn’t patched yet.

duck.utils.threading.patch.patch_threading(*, pre_hook: Optional[Callable[[threading.Thread], None]] = None, post_hook: Optional[Callable[[threading.Thread], None]] = None, patch_existing_threads: bool = False) → None[source]

Monkey-patches threading.Thread so that:

  • New threads automatically track parent thread objects.

  • .run() is wrapped at instance level.

  • Subclass overrides continue to work.

  • Optionally patch existing Thread objects created before patching.

  • Automatic cleanup in registry.

Parameters:

  • pre_hook – runs before each thread’s original run().

  • post_hook – runs after each thread’s run() completes.

  • patch_existing_threads – If True, already-created threads that have NOT started will get their .run() patched as well.

Notes:

  • Idempotent: calling twice does nothing.

  • Back-patching cannot discover parent thread for already-created threads.

duck.utils.threading.patch.thread_info: Dict[int, Dict[str, Any]]

None