Last active
April 28, 2024 10:34
-
-
Save luis261/9c663b3ce7e3ec0436cb343ecfd8e394 to your computer and use it in GitHub Desktop.
metaprogrammable dicts in Python: dynamically define custom, (optionally) recursive dictionary types. Load in/generate only the wanted additional validation/general functionality in a declarative manner; allows fine-tuning via inversion of control
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from .common import VHandlingError as ValHandlingError | |
from .boundary import VHandlingIn as ValHandlingInterface | |
from .core import EnhancedDictFactory | |
from .proto_factory import ProtoFactory | |
__all__ = [ | |
"ValHandlingInterface", | |
EnhancedDictFactory.__name__ | |
] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from functools import wraps | |
""" decorators """ | |
def _autofmt(to_deco): | |
@wraps(to_deco) | |
def fmt_wrapper(*args, msg=None, **kwargs): | |
assert not msg is None | |
return to_deco(*args, **{"msg": msg.format(*_arg_fmt(args)), **kwargs}) | |
return fmt_wrapper | |
def _parameterized_1st(to_wrap): | |
def outer(client_code_cfg_param): | |
@wraps(to_wrap) | |
def inner(*args, **kwargs): | |
return to_wrap(client_code_cfg_param, *args, **kwargs) | |
return inner | |
return outer | |
def _add_handling_pass_cfg(*exctypes, err_hdl=None): | |
def _add_handling_inner(to_wrap): | |
@wraps(to_wrap) | |
def wrapper(*args, **kwargs): | |
try: | |
return to_wrap(*args, **kwargs) | |
except tuple(exctypes): | |
return None if err_hdl is None else err_hdl(*args, **kwargs) | |
return wrapper | |
return _add_handling_inner | |
""" functions """ | |
def _arg_fmt(args): | |
return [repr(str(a)) for a in args] | |
def _merge_dicts(a, b): | |
res = type(a)(**a, **b) | |
if len(res) < len(a) + len(b): | |
raise KeyError("Detected conflicting keys!") | |
else: | |
return res | |
def _is_ty_else_raise(t): | |
def _no_conv_tcheck(args): | |
if not isinstance(args[0], t): | |
raise TypeError("Expected object of type: " + repr(t) | |
+ ", got: " + repr(type(args[0]))) | |
return args | |
return _no_conv_tcheck |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from enum import Enum | |
from .common import _VHDirectives | |
class VHandlingIn(Enum): | |
"""Accepts code/cfg to tie into lib-internal constructs in a controlled way.""" | |
PASS = _VHDirectives._pass | |
WARN = _VHDirectives._warn | |
CONVERT = _VHDirectives._convert | |
ABORT = _VHDirectives._abort | |
RAISE = _VHDirectives._raise |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from functools import wraps | |
import warnings | |
from ._utils import _arg_fmt, _autofmt, _merge_dicts, _parameterized_1st | |
class AccumError(Exception): | |
"""Exception to be raised on unsound or failed accumulations.""" | |
pass | |
class VHandlingError(Exception): | |
"""This gets raised on events as prescribed by the client code.""" | |
pass | |
class VHandlingAbort(Exception): | |
"""Exception for propagating aborts as prescribed by the client.""" | |
pass | |
""" accumulator decorators """ | |
def merge_if_dict(acc): | |
@wraps(acc) | |
def wrapper(existing, new): | |
if isinstance(existing, dict): | |
return _merge_dicts(existing, new) | |
return acc(existing, new) | |
return wrapper | |
def to_accerr(etype): | |
def accerr_deco(to_wrap): | |
@wraps(to_wrap) | |
def handle_to_accerr(existing, new): | |
try: | |
return to_wrap(existing, new) | |
except etype as e: | |
raise AccumError from e | |
return handle_to_accerr | |
return accerr_deco | |
def no_accum(existing, new): | |
raise AccumError | |
nested_accum_only = to_accerr(KeyError)(merge_if_dict(no_accum)) | |
def conv_1st_to(t): | |
def dict_else_convert(args): | |
if isinstance(args[0], dict): | |
raise TypeError | |
args[0] = t(args[0]) | |
return args | |
return dict_else_convert | |
def dedup_naive(x): # no preservation of order, does not work for unhashable types | |
if isinstance(x, dict): | |
return x | |
return type(x)(set(x)) | |
class _VHDirectives(): | |
@staticmethod | |
def _pass(*args, lmbd=None, **kwargs): | |
return args[0] if lmbd is None else lmbd(*args) | |
@_autofmt | |
@staticmethod | |
def _warn(*args, msg=None, **kwargs): | |
warnings.warn("Detected " + msg, RuntimeWarning) | |
return _VHDirectives._pass(*args, **kwargs) | |
@_parameterized_1st | |
@staticmethod | |
def _convert(conversion_func, *args, msg=None, **kwargs): | |
assert not conversion_func is None | |
assert not msg is None | |
msg = msg.format(*_arg_fmt(args)) | |
return _VHDirectives._pass(*conversion_func(args), **kwargs) | |
@_autofmt | |
@staticmethod | |
def _abort(*args, msg=None, **kwargs): | |
warnings.warn("Prevented " + msg, RuntimeWarning) | |
raise VHandlingAbort() | |
@_autofmt | |
@staticmethod | |
def _raise(*args, msg=None, **kwargs): | |
raise VHandlingError("Failed on " + msg) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from copy import copy | |
from functools import partial, wraps | |
from ._utils import _add_handling_pass_cfg as _add_handling | |
from .common import AccumError, VHandlingAbort | |
def _autofill_args_wrap_seti(to_wrap): | |
@wraps(to_wrap) | |
def capture_factory_self(factory, *args, **kwargs): | |
if (factory._CFG_COMPOSITION_ORDER.index(to_wrap.__name__) <= | |
factory._successive_composition_idx): | |
raise TypeError( | |
"Can't call \"" + to_wrap.__name__ | |
+ "\" now, please respect the enforced configuration order by calling it earlier!" | |
) | |
else: | |
factory._successive_composition_idx = factory._CFG_COMPOSITION_ORDER.index(to_wrap.__name__) | |
if to_wrap.__name__ == "nested": | |
factory._name_cfg = "Nested" + factory._name_cfg | |
elif not factory._CFG_METH_TO_NAME_SUFFIX.get(to_wrap.__name__) is None: | |
factory._name_cfg += factory._CFG_METH_TO_NAME_SUFFIX[to_wrap.__name__] | |
else: | |
pass | |
if to_wrap.__name__ == "on_overrides": | |
factory._override_dir_given = True | |
def seti_wrapper( | |
itself, key, to_insert, | |
inner=factory._nspdict_cfg["__setitem__"], **_internal_prio | |
): | |
return to_wrap(inner, itself, key, to_insert, *args, **{**kwargs, **_internal_prio}) | |
factory._nspdict_cfg["__setitem__"] = seti_wrapper | |
return factory # enables meth chaining for all decorated methods | |
return capture_factory_self | |
class EnhancedDictFactory(): | |
""" | |
Core factory to dynamically generate code for custom `dict` types. | |
All arguments for methods decorated with `_autofill_args_wrap_seti` | |
starting with an underscore shall be omitted by calls from client code. | |
""" | |
_CFG_COMPOSITION_ORDER = [ | |
"on_overrides", | |
"with_keys_of", | |
"nested", | |
"on_inputs" | |
] | |
_CFG_METH_TO_NAME_SUFFIX = { | |
"on_overrides": "WithCustomOverrides", | |
"with_keys_of": "WithTypeCheckedKeys", | |
"on_inputs": "WithCustomInputHandling" | |
} | |
def __init__(self): | |
self.bases_cfg = (dict,) | |
self._name_cfg = "EnhancedCustomDict" | |
self._successive_composition_idx = -1 | |
self._override_dir_given = False | |
def _bypass_accum(itself, key, to_insert, _handler=None): | |
# no need for cooperative multiple inheritance here, would | |
# rather directly depend on the exact implementation we want | |
# instead of relying on `super` stackframe access compiler magic | |
dict.__setitem__(itself, key, | |
(_handler or itself)._handle_duplicates(to_insert, key)) | |
return to_insert | |
def _seti_basecase(itself, key, to_insert, _handler=None): | |
""" | |
Perform the innermost insert without bypassing accumulation. | |
The `_handler` kwarg is useful when an item should be inserted | |
on a "dumb" type which can't handle accumulation/duplicates; | |
in that case you can pass in an instance which has the additional | |
capabilites as `_handler`, while passing the actual instance to | |
insert on as usual via `itself`. | |
""" | |
try: | |
itself[key] | |
except KeyError: # nonexistent key | |
pass | |
else: | |
to_insert = (_handler or itself)._accum(itself[key], to_insert) | |
return _bypass_accum(itself, key, to_insert, _handler=_handler) | |
self._nspdict_cfg = { | |
"__setitem__": _seti_basecase, | |
"_dupl_dir": None, | |
"_bypass_accum": _bypass_accum, | |
"_handle_duplicates": self.__class__._handle_duplicates | |
} | |
@staticmethod | |
def from_proto(proto, cfg_stash=None): | |
"""Extract a factory managed in a `ProtoFactory` by deploying it.""" | |
if not cfg_stash is None: | |
for cfg_meth_name in cfg_stash: # optional, adhoc application | |
# (preferably handled upfront, directly via the proto instance instead) | |
getattr(proto, cfg_meth_name)(cfg_stash[cfg_meth_name]) | |
return proto.deploy() | |
def build(self, accum=None): | |
""" | |
Dynamically create a class inheriting from `dict`. | |
The returned `dict` class is augmented according to any | |
config applied previously via the other factory methods. | |
If you directly supply an accumulator (`accum` kwarg), | |
it overrides any previously configured accumulator, | |
but only for this particular build call. | |
If no accumulator is passed nor was previously configured, | |
a hardcoded fallback is used. | |
""" | |
# can't operate inplace to avoid sideffects but don't need/can't use deepcopy | |
out_nsp = copy(self._nspdict_cfg) | |
if not self._override_dir_given: | |
type(self)._wrap_seti( | |
out_nsp, | |
_add_handling(AccumError, # general handling of accum failure signals | |
err_hdl=self._nspdict_cfg["_bypass_accum"] # fallback on bypass | |
) | |
) | |
# aborts should stop the current call | |
# but we don't want them popping up in the client code | |
type(self)._wrap_seti(out_nsp, _add_handling(VHandlingAbort)) | |
out_nsp["_accum"] = staticmethod( | |
(accum or self._nspdict_cfg.get("_accum", None) | |
or (lambda existing,new:existing+new)) | |
) | |
return type(self._name_cfg, self.bases_cfg, out_nsp) | |
def __call__(self, *args, **kwargs): | |
"""Shortcut for more explicit call via `build`.""" | |
return self.build(*args, **kwargs) | |
def with_custom_acc(self, accum): | |
"""Configure a custom accumulator.""" | |
self._nspdict_cfg["_accum"] = accum | |
return self | |
def on_duplicates( | |
self, handling_directive, dedup_criterion, | |
is_dupl=lambda x:len(set(x))<len(x) | |
): | |
"""Specify code to run on duplicates, as specified by the criteria.""" | |
self._name_cfg += "WithCustomHandlingOnDupl" | |
self._nspdict_cfg["_dupl_dir"] = staticmethod(handling_directive) | |
self._nspdict_cfg["_dedup_crit"] = staticmethod(dedup_criterion) | |
self._nspdict_cfg["_is_dupl"] = staticmethod(is_dupl) | |
return self | |
@_autofill_args_wrap_seti | |
def on_overrides( | |
_base_fn, _itself, _key, _to_insert, | |
handling_directive, _handler=None | |
): | |
"""Configure a handling directive that gets invoked on overrides.""" | |
try: | |
return _base_fn(_itself, _key, _to_insert, _handler=_handler) | |
except AccumError: | |
return handling_directive( | |
_to_insert, _itself[_key], | |
msg="override of {1} with {0} under: " + repr(_key), | |
lmbd=lambda i, _ : partial(type(_handler or _itself)._bypass_accum, _itself, _key)(i, _handler=_handler) | |
) | |
@_autofill_args_wrap_seti | |
def with_keys_of( | |
_base_fn, _itself, _key, _to_insert, | |
*args, _handler=None | |
): | |
"""Make the resulting class only accept keys of a certain type.""" | |
if not isinstance(_key, tuple(args)): | |
raise KeyError( | |
"Expected any of types: " + repr(tuple(args)) | |
+ ", got key: " + repr(_key) | |
+ " of type " + repr(type(_key)) + " instead" | |
) | |
return _base_fn(_itself, _key, _to_insert, _handler=_handler) | |
@_autofill_args_wrap_seti | |
def nested(_base_fn, _itself, _key, _to_insert, custom_subtype=None): | |
"""Generate code to enable nested access via tuple notation.""" | |
if isinstance(_key, tuple): | |
if len(_key) > 1: | |
ty = custom_subtype or type(_itself) | |
sub = _itself | |
for subcat in _key[:-1]: | |
try: | |
sub = _base_fn(sub, subcat, ty(), _handler=_itself) | |
except AccumError: # if no override directive configured | |
sub = type(_itself)._bypass_accum( | |
sub, subcat, ty(), _handler=_itself | |
) | |
return _base_fn(sub, _key[-1], _to_insert, _handler=_itself) | |
elif len(_key) == 1: | |
_key = _key[0] | |
return _base_fn(_itself, _key, _to_insert) | |
@_autofill_args_wrap_seti | |
def on_inputs(_base_fn, _itself, _key, _to_insert, handling_directive): | |
"""Configure code wrapped in handling interface to be run on all inputs.""" | |
return _base_fn( | |
_itself, _key, | |
handling_directive(_to_insert, msg="input {0} for: " + repr(_key)) | |
) | |
def _handle_duplicates(itself, val, key): | |
if (not itself._dupl_dir is None and type(itself)._dedup_crit(val) | |
and type(itself)._is_dupl(val)): | |
return itself._dupl_dir(val, | |
msg="duplicate values: {0} for: " + repr(key)) | |
return val | |
@staticmethod | |
def _wrap_seti(nsp_dict, deco): | |
nsp_dict["__setitem__"] = deco(nsp_dict["__setitem__"]) | |
return nsp_dict |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Not perfect, should handle most cases | |
# while outperforming alternatives; | |
# be sure to inherit appropriately in your custom types. | |
from collections.abc import Iterable | |
from functools import reduce, wraps | |
from ._utils import _is_ty_else_raise | |
from .boundary import VHandlingIn | |
from .core import EnhancedDictFactory | |
from .common import ( | |
AccumError, | |
conv_1st_to as conv_to, | |
merge_if_dict, | |
nested_accum_only, | |
no_accum, | |
to_accerr | |
) | |
""" private, local utils """ | |
def _pass_instance_to_mod_else_new(to_wrap): | |
@wraps(to_wrap) | |
def wrapper(instance=None): | |
return to_wrap(instance or EnhancedDictFactory()) | |
return wrapper | |
""" defaults """ | |
def sane_dedup_criterion(x): | |
return (isinstance(x, Iterable) | |
and not isinstance(x, | |
(dict, set, str))) | |
def dedup_preserve_order(to_dedup): # n^2 | |
return type(to_dedup)( | |
reduce( | |
lambda li, x: li if x in li else li.append(x) or li, | |
to_dedup, [] | |
) | |
) | |
def sane_iter_handling(acc): | |
@wraps(acc) | |
def wrapper(existing, new): | |
if (not isinstance(existing, Iterable) | |
or isinstance(existing, str)): | |
raise AccumError | |
return acc(existing, new) | |
return wrapper | |
def gen_accum(inner=lambda existing,new:existing+new): | |
return to_accerr(KeyError)( | |
merge_if_dict(sane_iter_handling( | |
to_accerr(TypeError)(inner) | |
)) | |
) | |
sane_accum = gen_accum() | |
conv_to_set = VHandlingIn.CONVERT(conv_to(set)) | |
tcheck_no_conv = lambda t:VHandlingIn.CONVERT(_is_ty_else_raise(t)) | |
# You can use the `with_*` functions below as starting points for your own types. | |
# When doing so, you'll likely run into type errs due to invalid call ordering. | |
# To circumvent that, instead of directly passing an `EnhancedDictFactory`, | |
# try passing in a `ProtoFactory` with build forwarding turned off as the | |
# `instance` keyword argument. This way, the with-fn can apply its defaults on | |
# the proto object which acts as a proxy. You can then add to (or even override) | |
# that base setup, the resulting config only gets applied once you call `deploy` | |
# on the proxy. You will then receive the actual, internal factory, | |
# configured as specified. | |
""" | |
`dict` class creation functions | |
utilize the instance kwarg for access to | |
the object which generated the output class | |
""" | |
@_pass_instance_to_mod_else_new | |
def with_min_defaults(factory_instance): | |
"""Work like a `builtins.dict`, but restrict keys to `str`.""" | |
return factory_instance.with_custom_acc(no_accum).with_keys_of(str).build() | |
@_pass_instance_to_mod_else_new | |
def with_sane_dedup(factory_instance): | |
return factory_instance.with_custom_acc(no_accum).on_duplicates( | |
VHandlingIn.CONVERT(lambda args:(dedup_preserve_order(args[0]),)), | |
sane_dedup_criterion | |
).build() | |
@_pass_instance_to_mod_else_new | |
def with_sane_defaults(factory_instance): | |
with_sane_dedup(instance=factory_instance) | |
return factory_instance.with_custom_acc( | |
nested_accum_only | |
).on_overrides( | |
VHandlingIn.WARN | |
).with_keys_of(str).nested( | |
custom_subtype=dict | |
).build() | |
@_pass_instance_to_mod_else_new | |
def with_nice_defaults(factory_instance): | |
with_sane_dedup(instance=factory_instance) | |
factory_instance.with_custom_acc(sane_accum) | |
factory_instance.on_overrides(VHandlingIn.ABORT) | |
factory_instance.with_keys_of(str) | |
factory_instance.nested() | |
return factory_instance.build() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
{ | |
"metadata": { | |
"kernelspec": { | |
"name": "python", | |
"display_name": "Python (Pyodide)", | |
"language": "python" | |
}, | |
"language_info": { | |
"codemirror_mode": { | |
"name": "python", | |
"version": 3 | |
}, | |
"file_extension": ".py", | |
"mimetype": "text/x-python", | |
"name": "python", | |
"nbconvert_exporter": "python", | |
"pygments_lexer": "ipython3", | |
"version": "3.8" | |
} | |
}, | |
"nbformat_minor": 4, | |
"nbformat": 4, | |
"cells": [ | |
{ | |
"cell_type": "code", | |
"source": "def _panic():\n assert False\n\n\nfrom dict_manufacturing import (\n ValHandlingError,\n ValHandlingInterface,\n EnhancedDictFactory,\n ProtoFactory,\n defaults\n)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 1 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Let's start off with an example which is focused on quickly demonstrating some of the key features this package provides. Keep reading for an explanation which starts off with more basic usage and progressively introduces new constructs while explaining them.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "CustomDictClass = EnhancedDictFactory().on_overrides(\n ValHandlingInterface.ABORT # once a key is assigned, we don't allow it to be overridden\n).with_keys_of(str).nested().on_duplicates(\n ValHandlingInterface.WARN, # emit warnings on values which contain duplicate items\n lambda x:isinstance(x, list) # only check values of type `list` for duplicates\n).build(accum=defaults.no_accum) # no combination of existing and new values on assignments to existing keys", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 2 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "You can use the instance of the custom type like you would a builtin `dict`, but the additional constraints and validation rules apply.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "custom = CustomDictClass()\ncustom[\"a\"] = 1", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 3 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "We could also start off our custom type by using a default applying function and then altering the factory until it suits our needs.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "de_factory = EnhancedDictFactory()\n# Wrap our factory, which means we can still override config applied in the next step.\nwrapper = ProtoFactory(instance=de_factory, fwd_build=False)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 4 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "defaults.with_sane_defaults(instance=wrapper)\n# Let's say we're completely fine with overrides and want to avoid the warnings,\n# which would get emitted per the sane default function.\n# Because we passed in the wrapper, we're free to overrule that behavior.\nwrapper.on_overrides(ValHandlingInterface.PASS)\n# Next, let's deploy the factory:\n# You don't need to keep a reference around,\n# it is extracted on deployment.\nif not de_factory is wrapper.deploy():\n _panic()", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 5 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "AnotherCustomDiType = de_factory.build()", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 6 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "For entirely premade types, check the `prefab` module. For preconfigured factories and other config-related stuff, check the `defaults` module.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder = EnhancedDictFactory() # instantiate factory", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 7 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Create class without passing any config beforehand for demonstration purposes. This is NOT recommended.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "Base = builder.build()", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 8 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Instantiate the custom class, it works much like a `dict`.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "basic_instance = Base()", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 9 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "basic_instance[\"a\"] = 1\nbasic_instance[\"b\"] = 2\nprint(basic_instance)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': 1, 'b': 2}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 10 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Reassigning a key produces unexpected results:", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "basic_instance[\"a\"] = 1", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 11 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "`basic_instance[\"a\"]` is now equal to `2`. The reason being that in the absence of a custom accumulator, we're running a fallback implementation; on a reassignment, it simply tries to naively add the new value to the existing one. This is neither practical nor safe, since it could produce a `TypeError` depending on the input, which would not get handled.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "print(basic_instance)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': 2, 'b': 2}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 12 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Accumulation is a feature usually only meant for Iterables.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Let's define an accumulator more suited to our needs. It raises a custom error which lets the invocating code know an accumulation is not desired/couldn't be carried out. An assignment (override) is carried out instead for such cases.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "from collections.abc import Iterable\ndef sensible_acc(existing, new):\n if isinstance(existing, Iterable):\n try:\n return existing + new\n except TypeError:\n raise defaults.AccumError(\"The two types do not fit together, please override with the new value instead.\")\n else:\n raise defaults.AccumError(\"Accumulation not appropriate, please override with the new value instead.\")", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 13 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "This time, when building the class, we pass in our custom accumulator.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "sensible_dict = builder.build(accum=sensible_acc)() # create instance with desired behavior inline\nsensible_dict[\"a\"] = 1", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 14 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Try assigning to an existing key.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "sensible_dict[\"a\"] = 1", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 15 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "`a` is now still `1`, despite the additional call.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "print(sensible_dict)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': 1}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 16 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "When we reassign to a different number, we obviously still get the expected result.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "sensible_dict[\"a\"] = 4\nprint(sensible_dict)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': 4}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 17 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "On values that don't result in an `AccumError`, the addition still runs on assignment.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "sensible_dict[\"c\"] = [2, 3]\nsensible_dict[\"c\"] = [4]\nprint(sensible_dict)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': 4, 'c': [2, 3, 4]}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 18 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "> Hint: an accumulator roughly equivalent to our custom function but which offers more broad applicability is defined in the `defaults` module:\n\n`from dict_manufacturing.defaults import sane_accum`", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Configure acc in persistent manner, so we don't have to pass it on every `build` call.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder.with_custom_acc(defaults.sane_accum)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"execution_count": 19, | |
"output_type": "execute_result", | |
"data": { | |
"text/plain": "<dict_manufacturing.core.EnhancedDictFactory at 0x1354c58>" | |
}, | |
"metadata": {} | |
} | |
], | |
"execution_count": 19 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "Subscribe the following handling directive to any \"override\" events; once triggered, it aborts the offending call and raises a custom error.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder.on_overrides(ValHandlingInterface.RAISE)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"execution_count": 20, | |
"output_type": "execute_result", | |
"data": { | |
"text/plain": "<dict_manufacturing.core.EnhancedDictFactory at 0x1354c58>" | |
}, | |
"metadata": {} | |
} | |
], | |
"execution_count": 20 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder.with_keys_of(str) # only allow strings for keys", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"execution_count": 21, | |
"output_type": "execute_result", | |
"data": { | |
"text/plain": "<dict_manufacturing.core.EnhancedDictFactory at 0x1354c58>" | |
}, | |
"metadata": {} | |
} | |
], | |
"execution_count": 21 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "a_dict = builder.build()()\na_dict[\"a\"] = 1\ntry:\n a_dict[\"a\"] = 0\nexcept ValHandlingError: # no overrides allowed\n pass\nelse:\n _panic()\n\ntry:\n a_dict[1] = 1\nexcept KeyError: # only keys of type str are allowed\n pass\nelse:\n _panic()", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 22 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder.on_duplicates(\n ValHandlingInterface.WARN, # warn on duplicates\n defaults.sane_dedup_criterion\n)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"execution_count": 23, | |
"output_type": "execute_result", | |
"data": { | |
"text/plain": "<dict_manufacturing.core.EnhancedDictFactory at 0x1354c58>" | |
}, | |
"metadata": {} | |
} | |
], | |
"execution_count": 23 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "We can gain some performance by sacrificing convenience when using deep nesting. Now, you can only perform nested assignments via references to the top level dictionary instance. Without this adjustment, you can also set nested values via references to `dict` instances which are themselves a value under a certain key of a parent `dict`. This may be convenient, but causes additional overhead, because the nested values can't be raw `dict` instances, instead, they have to be of the same type as the top level dictionary.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "builder.nested(custom_subtype=dict)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"execution_count": 24, | |
"output_type": "execute_result", | |
"data": { | |
"text/plain": "<dict_manufacturing.core.EnhancedDictFactory at 0x1354c58>" | |
}, | |
"metadata": {} | |
} | |
], | |
"execution_count": 24 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "kv = builder.build()()\nkv[\"a\"] = [1, 2, 1] # this results in a duplicate warning", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stderr", | |
"text": "/drive/dict_manufacturing/common.py:70: RuntimeWarning: Detected duplicate values: '[1, 2, 1]' for: 'a'\n warnings.warn(\"Detected \" + msg, RuntimeWarning)\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 25 | |
}, | |
{ | |
"cell_type": "markdown", | |
"source": "The following notation does not extend to `__getitem__` or `__delitem__`, you'll have to use the more explicit default notation instead.", | |
"metadata": {} | |
}, | |
{ | |
"cell_type": "code", | |
"source": "kv[\"b\", \"a\"] = 1 # nested setting", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [], | |
"execution_count": 26 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "print(kv[\"b\"][\"a\"])", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "1\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 27 | |
}, | |
{ | |
"cell_type": "code", | |
"source": "kv[\"c\", \"a\", \"a\"] = 1\n\nkv[\"d\"] = 1\n\ntry:\n kv[\"d\", \"a\"] = 1 # this raises because `\"d\"` is already set\nexcept ValHandlingError:\n pass\nelse:\n _panic()\n\nprint(kv)", | |
"metadata": { | |
"trusted": true | |
}, | |
"outputs": [ | |
{ | |
"name": "stdout", | |
"text": "{'a': [1, 2, 1], 'b': {'a': 1}, 'c': {'a': {'a': 1}}, 'd': 1}\n", | |
"output_type": "stream" | |
} | |
], | |
"execution_count": 28 | |
} | |
] | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from .boundary import VHandlingIn | |
from .common import conv_1st_to, dedup_naive, no_accum, nested_accum_only | |
from .core import EnhancedDictFactory | |
_gen_base = lambda:EnhancedDictFactory().with_custom_acc(no_accum) | |
SetValuesDict = _gen_base().on_inputs( | |
VHandlingIn.CONVERT( | |
conv_1st_to(set)) | |
).build() | |
BasicNestedDict = EnhancedDictFactory().nested( | |
custom_subtype=dict).build(accum=nested_accum_only) | |
FrozenAssignmentsDict = _gen_base().on_overrides(VHandlingIn.ABORT).build() | |
NaiveDedupValuesDict = _gen_base().on_duplicates( | |
VHandlingIn.CONVERT( | |
lambda *args:dedup_naive(args[0]) | |
), lambda _:True, is_dupl=lambda:True | |
).build() | |
IntersectedSetValuesDict = EnhancedDictFactory().on_overrides( | |
VHandlingIn.RAISE | |
).on_inputs( | |
VHandlingIn.CONVERT(conv_1st_to(frozenset)) | |
).build(accum=set.intersection) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from copy import deepcopy | |
import warnings | |
from .core import EnhancedDictFactory | |
class ProtoFactory(): | |
""" | |
An alternative for passing raw factories. | |
Think of this class as a proxy to route method calls on a factory | |
in a time indirective manner which ensures they are ultimately | |
applied in the mandated order. | |
In cases where you want to configure a factory without adhering | |
to the call order enforced by raw factory objects, you can use | |
this as a stand-in replacement over a conventional factory. | |
This can be especially useful when passing the object through | |
a `with_*` function from the `defaults` module, which may be used | |
as a point to start of your own definition. By opting for an object | |
of this type, you avoid hard committing to the applied defaults | |
and retain the ability to perform overrides after the fact, | |
before ultimately deploying the type. | |
""" | |
def __init__(self, fwd_build=True, instance=None): | |
self._fwd_build = fwd_build | |
self._frozen = False | |
self._stash = {} | |
# composition over inheritance, specifically here to circumvent requiring __getattribute__ | |
self._factory = instance or EnhancedDictFactory() | |
def __getattr__(self, attr): | |
r"""Attribute access interception "lite".""" | |
if self._frozen: | |
raise AttributeError("Already configured downstream factory, permanently detached delayed attribute forwarding for this instance!") | |
def cfg_applier(*args, **kwargs): | |
self._stash[attr] = { | |
"args": args, | |
"kwargs": kwargs | |
} | |
return self # sustain chaining feature | |
return cfg_applier | |
def clone(self): | |
""" | |
Acquire a copy of this factory wrapper. | |
This can help avoid redundant reiteration of config | |
when defining multiple types with definitions that all | |
share a common set of basic configuration. | |
You obviously could define your own function instead, | |
but having the ability to clone at various points | |
throughout the definition process is more flexible. | |
""" | |
return deepcopy(self) | |
def build(self, *args, **kwargs): | |
""" | |
Create the custom class, via the internal factory. | |
Only if `_fwd_build` is truthy, does the config actually get applied. | |
""" | |
if self._fwd_build: | |
return self.deploy().build() | |
def __call__(self, *args, **kwargs): | |
"""Call config application via less explicit alias.""" | |
return self.deploy(*args, **kwargs) | |
def deploy(self): | |
"""Apply cached config calls to the managed factory and return it.""" | |
if self._frozen: | |
warnings.warn("The config which was cached here has already been passed onwards, this instance is depleted! Perhaps you called this repeatedly by mistake?", RuntimeWarning) | |
return self._factory | |
self._frozen = True | |
ordered_cfg = [None for _ in range(len(self._factory._CFG_COMPOSITION_ORDER))] | |
for attr, cfg_args in self._stash.items(): | |
try: | |
ordered_cfg[self._factory._CFG_COMPOSITION_ORDER.index(attr)] = cfg_args | |
except ValueError: | |
self._propagate(attr, cfg_args) | |
for idx, cfg in enumerate(ordered_cfg): | |
if cfg is None: | |
continue | |
self._propagate(self._factory._CFG_COMPOSITION_ORDER[idx], cfg) | |
del self._stash | |
return self._factory | |
def _propagate(self, attr, cfg): | |
return getattr(self._factory, attr)( | |
*cfg["args"], **cfg["kwargs"] | |
) | |
def __deepcopy__(self, _): | |
warnings.warn("Deepcopy support for this type is tentative", PendingDeprecationWarning) | |
if self._frozen: | |
warnings.warn("You are cloning an object which references an already configured factory," | |
+ " which won't be copied to the clone", RuntimeWarning) | |
pf = self.__class__() | |
pf._factory = EnhancedDictFactory() | |
pf._fwd_build = self._fwd_build | |
pf._frozen = self._frozen | |
pf._stash = deepcopy(self._stash) | |
return pf |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment