feat: add cdf invoke function to implicitly use conf and di from work…

…space when active

src/cdf/__init__.py

@@ -8,6 +8,7 @@
 )
 from cdf.core.context import (
     get_active_workspace,
+    invoke,
     resolve,
     set_active_workspace,
     use_workspace,
@@ -31,5 +32,6 @@
     "get_active_workspace",
     "set_active_workspace",
     "resolve",
+    "invoke",
     "use_workspace",
 ]

src/cdf/core/context.py

@@ -2,13 +2,16 @@
 
 import contextlib
 import functools
+import logging
 import typing as t
 from contextvars import ContextVar, Token
 
 if t.TYPE_CHECKING:
     from cdf.core.injector import Lifecycle
     from cdf.core.workspace import Workspace
 
+logger = logging.getLogger(__name__)
+
 _ACTIVE_WORKSPACE: ContextVar[t.Optional["Workspace"]] = ContextVar(
     "active_workspace", default=None
 )
@@ -93,6 +96,16 @@ def wrapper(*args: t.Any, **kwargs: t.Any) -> T:
     return _resolve
 
 
+def invoke(func_or_cls: t.Callable, *args: t.Any, **kwargs: t.Any) -> t.Any:
+    """Invoke a function or class with resolved dependencies."""
+    workspace = get_active_workspace()
+    if workspace is None:
+        logger.debug("Invoking %s without a bound workspace", func_or_cls)
+        return func_or_cls(*args, **kwargs)
+    logger.debug("Invoking %s bound to workspace %s", func_or_cls, workspace)
+    return workspace.invoke(func_or_cls, *args, **kwargs)
+
+
 def get_default_callable_lifecycle() -> t.Optional["Lifecycle"]:
     """Get the default lifecycle for callables when otherwise unspecified."""
     return _DEFAULT_CALLABLE_LIFECYCLE.get()