Skip to content

Commit

Permalink
release v3.77.20190
Browse files Browse the repository at this point in the history
  • Loading branch information
@klahnakoski
klahnakoski committed Jul 8, 2020
2 parents b63617e + b869841 commit e2dd7a5
Show file tree
Hide file tree
Showing 6 changed files with 280 additions and 121 deletions.
2 changes: 1 addition & 1 deletion .travis.yml
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
language: python
python:
- "2.7"
- "3.6"
- "3.7"

env:
global:
Expand Down
240 changes: 125 additions & 115 deletions mo_kwargs/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,19 @@

from functools import update_wrapper

from mo_dots import get_logger, is_data, wrap, zip as dict_zip, set_default
from mo_dots import get_logger, is_data, to_data, zip as dict_zip, set_default
from mo_future import (
get_function_arguments,
get_function_defaults,
get_function_name,
text,
)
is_text)
from mo_logs import Except

KWARGS = str("kwargs")


def override(func):
def override(kwargs=None):
"""
THIS DECORATOR WILL PUT ALL PARAMETERS INTO THE `kwargs` PARAMETER AND
THEN PUT ALL `kwargs` PARAMETERS INTO THE FUNCTION PARAMETERS. THIS HAS
Expand All @@ -38,123 +38,133 @@ def override(func):
3) DEFAULT VALUES ASSIGNED IN FUNCTION DEFINITION
"""

func_name = get_function_name(func)
params = get_function_arguments(func)
if not get_function_defaults(func):
defaults = {}
else:
defaults = {
k: v
for k, v in zip(reversed(params), reversed(get_function_defaults(func)))
}

def raise_error(e, a, k):
packed = set_default(dict(zip(params, a)), k)
err = text(e)
e = Except.wrap(e)
if err.startswith(func_name) and (
"takes at least" in err
or "takes exactly " in err
or "required positional argument" in err
):
missing = [p for p in params if str(p) not in packed]
given = [p for p in params if str(p) in packed]
if not missing:
raise e
else:
get_logger().error(
"Problem calling {{func_name}}: Expecting parameter {{missing}}, given {{given}}",
func_name=func_name,
missing=missing,
given=given,
stack_depth=2,
cause=e,
)
raise e

if KWARGS not in params:
# ADDING A kwargs PARAMETER TO SOME REGULAR METHOD
def wo_kwargs(*args, **kwargs):
settings = kwargs.get(KWARGS, {})
ordered_params = dict(zip(params, args))
a, k = params_pack(params, defaults, settings, kwargs, ordered_params)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(wo_kwargs, func)

elif func_name in ("__init__", "__new__") or params[0] in ("self", "cls"):

def w_bound_method(*args, **kwargs):
if len(args) == 2 and len(kwargs) == 0 and is_data(args[1]):
# ASSUME SECOND UNNAMED PARAM IS kwargs
a, k = params_pack(
params, defaults, args[1], {params[0]: args[0]}, kwargs
)
elif KWARGS in kwargs and is_data(kwargs[KWARGS]):
# PUT args INTO kwargs
a, k = params_pack(
params, defaults, kwargs[KWARGS], dict_zip(params, args), kwargs
)
else:
a, k = params_pack(params, defaults, dict_zip(params, args), kwargs)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(w_bound_method, func)
def output(func):
func_name = get_function_name(func)
params = get_function_arguments(func)
if not get_function_defaults(func):
defaults = {}
else:
defaults = {
k: v
for k, v in zip(reversed(params), reversed(get_function_defaults(func)))
}

def raise_error(e, a, k):
packed = set_default(dict(zip(params, a)), k)
err = text(e)
e = Except.wrap(e)
if err.startswith(func_name) and (
"takes at least" in err
or "takes exactly " in err
or "required positional argument" in err
):
missing = [p for p in params if str(p) not in packed]
given = [p for p in params if str(p) in packed]
if not missing:
raise e
else:
get_logger().error(
"Problem calling {{func_name}}: Expecting parameter {{missing}}, given {{given}}",
func_name=func_name,
missing=missing,
given=given,
stack_depth=2,
cause=e,
)
raise e

if kwargs not in params:
# ADDING A kwargs PARAMETER TO SOME REGULAR METHOD
def wo_kwargs(*given_args, **given_kwargs):
settings = given_kwargs.get(kwargs, {})
ordered_params = dict(zip(params, given_args))
a, k = params_pack(params, defaults, settings, given_kwargs, ordered_params)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(wo_kwargs, func)

elif func_name in ("__init__", "__new__") or params[0] in ("self", "cls"):

def w_bound_method(*given_args, **given_kwargs):
if len(given_args) == 2 and len(given_kwargs) == 0 and is_data(given_args[1]):
# ASSUME SECOND UNNAMED PARAM IS kwargs
a, k = params_pack(
params, defaults, given_args[1], {params[0]: given_args[0]}, given_kwargs
)
elif kwargs in given_kwargs and is_data(given_kwargs[kwargs]):
# PUT args INTO given_kwargs
a, k = params_pack(
params, defaults, given_kwargs[kwargs], dict_zip(params, given_args), given_kwargs
)
else:
a, k = params_pack(params, defaults, dict_zip(params, given_args), given_kwargs)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(w_bound_method, func)

else:
else:

def w_kwargs(*args, **kwargs):
if len(args) == 1 and len(kwargs) == 0 and is_data(args[0]):
# ASSUME SINGLE PARAMETER IS kwargs
a, k = params_pack(params, defaults, args[0])
elif KWARGS in kwargs and is_data(kwargs[KWARGS]):
# PUT args INTO kwargs
a, k = params_pack(
params, defaults, kwargs[KWARGS], dict_zip(params, args), kwargs
def w_kwargs(*given_args, **given_kwargs):
if len(given_args) == 1 and len(given_kwargs) == 0 and is_data(given_args[0]):
# ASSUME SINGLE PARAMETER IS kwargs
a, k = params_pack(params, defaults, given_args[0])
elif kwargs in given_kwargs and is_data(given_kwargs[kwargs]):
# PUT given_args INTO given_kwargs
a, k = params_pack(
params, defaults, given_kwargs[kwargs], dict_zip(params, given_args), given_kwargs
)
else:
# PULL kwargs OUT INTO PARAMS
a, k = params_pack(params, defaults, dict_zip(params, given_args), given_kwargs)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(w_kwargs, func)

def params_pack(params, *args):
"""
:param params:
:param args:
:return: (args, kwargs) pair
"""
settings = {}
for a in args:
for k, v in a.items():
settings[str(k)] = v
settings[kwargs] = to_data(settings)

if params and params[0] in ("self", "cls"):
s = settings.get(params[0])
if s is None:
return (
[],
{k: settings[k] for k in params[1:] if k in settings},
)
else:
# PULL kwargs OUT INTO PARAMS
a, k = params_pack(params, defaults, dict_zip(params, args), kwargs)
try:
return func(*a, **k)
except TypeError as e:
raise_error(e, a, k)

return update_wrapper(w_kwargs, func)


def params_pack(params, *args):
"""
:param params:
:param args:
:return: (args, kwargs) pair
"""
settings = {}
for a in args:
for k, v in a.items():
settings[str(k)] = v
settings[KWARGS] = wrap(settings)

if params and params[0] in ("self", "cls"):
s = settings.get(params[0])
if s is None:
return (
[],
{k: settings[k] for k in params[1:] if k in settings},
)
return (
[s],
{k: settings[k] for k in params[1:] if k in settings},
)
else:
return (
[s],
{k: settings[k] for k in params[1:] if k in settings},
[],
{k: settings[k] for k in params if k in settings}
)

if is_text(kwargs):
# COMPLEX VERSION @override(kwargs="other")
return output
elif kwargs == None:
raise NotImplementedError("use @override without calling")
else:
return (
[],
{k: settings[k] for k in params if k in settings}
)
# SIMPLE VERSION @override
func, kwargs = kwargs, KWARGS
return output(func)
Loading

0 comments on commit e2dd7a5

Please sign in to comment.