Iñigo Hidalgo
01/19/2023, 9:17 AMtest_size
need to be passed by name. It would need to be a combination of passing an iterable as well as a dictionary to the inputs
for the node, which as far as I know isn't doable. If not possible, how would you suggest I proceed, when my objective is to be able to feed in outputs from different nodes to converge into that function to then output into a train node.datajoely
01/19/2023, 9:45 AMfunctools.partial
to pass a literal python value to a node input you could pre-prepare your arguments that sort of way
https://stackoverflow.com/a/58875821/2010808Ben Horsburgh
01/19/2023, 9:47 AMIñigo Hidalgo
01/19/2023, 9:48 AMdatajoely
01/19/2023, 9:56 AMBen Horsburgh
01/19/2023, 10:31 AMimport inspect
from abc import ABC, abstractmethod
from typing import Tuple, Dict, Callable, Optional, List
import wrapt
class PositionalWithKwargs:
def __init__(self, convert_to_positional: List[str]):
self.convert_to_positional = convert_to_positional
def rewrite(self, *args, **kwargs) -> Tuple[Tuple, Dict]:
# extract positional from kwargs
positional = [kwargs[k] for k in self.convert_to_positional]
# remove positional from kwargs
kwargs = {
k: v for k, v in kwargs.items() if k not in self.convert_to_positional
}
# prepend args with positional
args = [*positional, *args]
return tuple(args), kwargs
def argspec_factory(self, wrapped) -> Optional[inspect.FullArgSpec]:
# 'args' is a list of the parameter names.
# 'varargs' and 'varkw' are the names of the * and ** parameters or None.
# 'defaults' is an n-tuple of the default values of the last n parameters.
# 'kwonlyargs' is a list of keyword-only parameter names.
# 'kwonlydefaults' is a dictionary mapping names from kwonlyargs to defaults.
# 'annotations' is a dictionary mapping parameter names to annotations.
(
args,
varargs,
varkw,
defaults,
kwonlyargs,
kwonlydefaults,
annotations,
) = inspect.getfullargspec(wrapped)
# grab arg defaults
defaults = defaults or ()
arg_defaults = {
arg: default
for arg, default in (
reversed(list(zip(reversed(args), reversed(defaults))))
)
}
# add positional args so they are expected as kwargs
kwonlyargs = kwonlyargs or []
kwonlydefaults = kwonlydefaults or {}
kwonlyargs.extend(self.convert_to_positional)
kwonlydefaults.update(
**{
arg: default
for arg, default in arg_defaults.items()
if arg in self.convert_to_positional
}
)
# remove positional args we now expect as kwargs
defaults = tuple(
default
for arg, default in (
reversed(list(zip(reversed(args), reversed(defaults))))
)
if arg not in self.convert_to_positional
)
args = [arg for arg in args if arg not in self.convert_to_positional]
return inspect.FullArgSpec(
args, varargs, varkw, defaults, kwonlyargs, kwonlydefaults, annotations
)
def __call__(self, func):
@wrapt.decorator(adapter=self.argspec_factory(func))
def __call__(wrapped, instance, args, kwargs):
args, kwargs = self.rewrite(*args, **kwargs)
return wrapped(*args, **kwargs)
return __call__(func)
def my_function(a, b, c, /, x):
return a + b + c + x
print(my_function(1, 2, 3, 4))
# >>> 10
wrapped_func = PositionalWithKwargs(["a", "b"])(my_function)
print(wrapped_func(1, a=2, b=3, x=4))
# >>> 10
fullargspec
- kedro uses this to map parameters correct, and users will use it anytime they read a stack trace or rely on IDE prompts
2. You need to intercept and rewrirte the args
and kwargs
sent to the call
3. You should use wrapt
as opposed to native python wrappers because they do a ton of amazing magic that preserved docstrings and the likesIñigo Hidalgo
01/19/2023, 11:20 AMdatajoely
01/19/2023, 11:27 AMBen Horsburgh
01/19/2023, 11:41 AM