On this page

• Package
  • Package
    • Package.cleanup()
    • Package.id
    • Package.unpack()
    • Package.unpack_and_cleanup()
    • Package.describe()
    • Package.is_coroutine()
    • Package.get_coroutine()
    • Package.is_async
    • Package.bundle()
    • Package.execute_sync()
    • Package.execute_async()
    • Package.verify()
    • Package.merge_many()
    • Package.bind_args()
    • Package.bind()
    • Package.override()
    • Package.curry()
    • Package.freeze()
    • Package.args
    • Package.kwargs
    • Package.signature

Package

Use md.Package from the public package namespace.

Implementation source

class Package(func: Callable[[...], R], *args: Any, **kwargs: Any)

cleanup() → None

Clean owned state in two phases and make the package unusable.

Contract:

• Idempotent: repeated calls return immediately.

• Marks _cleaned while the package lock is held before child cleanup begins.

• Performs reference nulling after the lock is released so core teardown does not occur inside the locked section.

Returns:

None.

property id: str

Returns a unique identifier for this Package instance. This is useful for tracking and debugging.

Returns:

A string representing the unique ID of the Package.

unpack() → Tuple[Callable[[...], Any], Tuple[Any, ...], Dict[str, Any]]

Deconstructs the Package into its core components.

This method provides a thread-safe way to get a snapshot of the original callable, its pre-bound positional arguments, and its pre-bound keyword arguments.

Returns:

• The original, unwrapped callable function.

• A tuple of the pre-bound positional arguments.

• A dictionary of the pre-bound keyword arguments.

Return type:

A tuple containing

unpack_and_cleanup() → Tuple[Callable[[...], Any], Tuple[Any, ...], Dict[str, Any]]

Destructively unpacks the Package and cleans it up in one atomic operation.

This method retrieves the original callable and its arguments, and then immediately disposes of the Package instance, making it unusable for any future operations. This is useful for "fire-and-forget" or "use-once" scenarios.

Returns:

• The original, unwrapped callable function.

• A tuple of the pre-bound positional arguments.

• A dictionary of the pre-bound keyword arguments.

Return type:

A tuple containing

Raises:

RuntimeError -- If the Package has already been cleaned.

describe() → str

Provides a human-readable description of the wrapped callable and its signature.

Contract:

• The signature text is the default inspect.signature rendering when every annotation name is available. An annotation naming a type that is unbound at runtime (a TYPE_CHECKING-only import under Python 3.14 lazy annotations) is rendered as its source text instead of raising.

• Callables without an inspectable signature render as (uninspectable).

Returns:

A string describing the function's name and parameters.

is_coroutine() → bool

Checks if the wrapped callable is a coroutine function.

Returns:

True if the wrapped item is a coroutine function, False otherwise.

Return type:

bool

get_coroutine() → Callable[[...], Awaitable[object]]

Returns the underlying coroutine function if the Package is async.

Returns:

The awaitable coroutine function.

Raises:

TypeError -- If the wrapped function is not a coroutine.

property is_async: bool

Check if the underlying function is async (coroutine). This is for info only.

Returns:

True if the original function was a coroutine function.

Return type:

bool

static bundle(item: Callable[[P], R] | Package[P, R] | Iterable[Callable[[P], R] | Package[P, R]] | None) → Package[P, R] | List[Package[P, R]]

Normalize one callable/package or an iterable of them into package objects.

Parameters:

item -- The input to normalize. Can be one callable, one Package, or an iterable containing a mix of both.

Returns:

Existing package for single-package input, new package for single-callable input, or a list of package objects for iterable input.

Return type:

Union["Package[P, R]", List["Package[P, R]"]]

Raises:

TypeError -- If item is None or contains invalid callable input.

execute_sync(*extra_args: ~P, **extra_kwargs: ~P) → R

Calls the wrapped function synchronously if it is sync.

This method is a convenience for invoking synchronous package objects. It gathers the arguments under the lock, then releases the lock before invoking the underlying function to avoid cross-thread contention when nested calls occur.

Returns:

The return value of the wrapped function.

Raises:

TypeError -- If the wrapped callable is asynchronous.

async execute_async(*extra_args: ~P, **extra_kwargs: ~P) → Any

Calls the wrapped function as a coroutine if it is async.

This method is a convenience for invoking asynchronous package objects. It gathers the arguments under the lock, then releases the lock before invoking the underlying function to avoid cross-thread contention when nested calls occur.

Returns:

The coroutine object returned by the wrapped function.

Raises:

TypeError -- If the wrapped callable is synchronous.

static verify(item_to_check: Any) → bool

Validate a package or iterable of packages.

This helper provides one boundary check for APIs that accept either one Package or a collection of Package objects.

Parameters:

item_to_check (Any) -- The item to validate. Can be one Package instance or an iterable containing package objects.

Returns:

Returns True if the validation succeeds.

Return type:

bool

Raises:

TypeError -- If the item_to_check (or any element within it) is not an instance of Package.

static merge_many(packs: Iterable[Package[..., Any]]) → Package[..., Any]

Pipe a sequence of Packages left-to-right into a single composite Package.

Example

combo = Package.merge_many([p1, p2, p3])
result = combo(x)   # approximately p3(p2(p1(x)))

bind_args(*new_args: Any) → Package[P, R]

Mutably replace the positional arguments of this Package.

Parameters:

*new_args -- New positional arguments to replace the current ones.

Returns:

self

Raises:

RuntimeError -- If the package is frozen.

bind(**new_kwargs: Any) → Package[P, R]

Mutably add or update keyword arguments.

Parameters:

**new_kwargs -- Keyword arguments to merge into the package.

Returns:

self

Raises:

RuntimeError -- If the package is frozen.

override(*args: Any, **kwargs: Any) → Package[P, R]

Mutably override both args and kwargs. WARNING: This modifies the original Package!

Parameters:

• *args -- Positional arguments to set.

• **kwargs -- Keyword arguments to merge.

Returns:

self

Raises:

RuntimeError -- If the package is frozen.

curry(*args: Any, **kwargs: Any) → Package[P, R]

Create a new Package with additional positional and keyword arguments.

Parameters:

• *args -- Extra args to append.

• **kwargs -- Extra kwargs to merge.

Returns:

A new Package with combined arguments.

freeze() → None

Prevent future argument mutation on this package instance.

Returns:

None.

property args: Tuple[Any, ...]

Return the stored positional arguments.

property kwargs: Dict

Return a copy of the stored keyword arguments.

A copy is returned (mirroring args) so external mutation cannot bypass bind()/freeze() and leave the cached signature stale.

property signature: Any

Return a pseudo-signature object representing bound args.

Returns:

Namespace whose arguments mapping contains positional placeholders (arg0, arg1, ...) plus keyword entries.

Return type:

SimpleNamespace

Topic reference · Full contents