-"""
-CRUDE stands for CRUD-Execution.
-It is a method to solve the problem of dealing with complex python objects in an
-environment that doesn't natively support these.
-
-The method's trick is to allow the complex object's that we "crudified" to be controlled
-via a string key that references the complex object, via a "store" which maps
-these string keys to the actual physical object.
-This store could be a python dictionary (so in RAM) or any persisting storage system
-(files, DB) that is given a `typing.Mapping` interface
-(see https://i2mint.github.io/dol/ or https://i2mint.github.io/py2store for
-tools to do so).
-
-Take, for instance, a GUI that allows a user to compute some descriptive statistics
-of the columns of a table.
-The inputs are a table, and one of the following statistics function:
-``statistics.mean``, ``statistics.median``, or ``statistics.stdev``.
-
-Python functions are not a type natively handled by GUI, so what can we do?
-We can stick a layer between our ``compute_stats(stats_func, table)`` function
-and our GUI, endowed with a
-``{"mean": statistics.mean``, "median": statistics.median, "std": statistics.stdev}``
-mapping. We expose the string keys to the GUI, and map them to the functions before
-calling ``compute_stats``.
-
-In the case of the ``table``, we'd probably add a means for the GUI user to upload
-tables (say from ``.csv`` or ``.xlsx`` files), storing them under a name of their
-choice, then pointing to said table via the name, when they want to execute a
-``compute_stats(stats_func, table)``.
-
-These are examples of what we call "crudifying" variables or functions.
-
-Here we therefore offer tools to do this sort of thing;
-wrap functions so that the complex arguments can be specified through a string key
-that points to the actual python object (which is stored in a session's memory or
-persisted in some fashion).
-"""
-
-from typing import Any, Literal, Mapping, Optional, Callable, Union, Iterable, Iterator
-from inspect import Parameter
-import os
-from functools import partial
-import time
-from numbers import Number
-
-from i2 import Sig, double_up_as_factory
-from i2.wrapper import Ingress, wrap
-from dol import Files, wrap_kvs
-from dol.filesys import mk_tmp_dol_dir, ensure_dir
-
-KT = str
-VT = Any
-StoreType = Mapping[KT, VT]
-StoreName = KT
-Mall = Mapping[StoreName, StoreType]
-
-
-
-
[docs]
-
def auto_key_from_arguments(*args, **kwargs) -> KT:
-
"""Make a str key from arguments.
-
-
>>> auto_key_from_arguments(1,2,c=3,d=4)
-
'1,2,c=3,d=4'
-
>>> auto_key_from_arguments(1,2)
-
'1,2'
-
>>> auto_key_from_arguments(c=3,d=4)
-
'c=3,d=4'
-
>>> auto_key_from_arguments()
-
''
-
"""
-
args_str = ','.join(map(str, args))
-
kwargs_str = ','.join(map(lambda kv: f'{kv[0]}={kv[1]}', kwargs.items()))
-
return ','.join(filter(None, [args_str, kwargs_str]))
-
[docs]
-
def auto_key_from_time(
-
*args, __format: Union[Number, str, Callable] = 1e6, **kwargs
-
) -> KT:
-
"""Make a str key with current timestamp (ignoring arguments)
-
-
:param __format: When a number, will be used as a multiplier of current utc time
-
-
>>> auto_key_from_time() # doctest: +SKIP
-
'1_669_724_787_630_906'
-
-
But `auto_key_from_time` is really meant to be used with ``functools.partial`` to
-
parametrize its ``__format``, such as:
-
-
>>> from functools import partial
-
>>>
-
>>> time_in_ms = partial(auto_key_from_time, __format=1e3)
-
>>> normal_format = partial(auto_key_from_time, __format='%Y-%m-%d %H:%M:%S')
-
>>> modulo_1000 = partial(auto_key_from_time, __format=lambda x: int(x % 1000))
-
>>>
-
>>> time_in_ms() # doctest: +SKIP
-
'1_669_724_787_641'
-
>>> normal_format() # doctest: +SKIP
-
'2022-11-29 12:26:27'
-
>>> modulo_1000() # doctest: +SKIP
-
'788'
-
"""
-
utc_seconds = time.time()
-
if isinstance(__format, Number):
-
return f'{int(utc_seconds * __format):_}'
-
elif isinstance(__format, str):
-
return time.strftime(__format, time.gmtime(utc_seconds))
-
else:
-
assert callable(
-
__format
-
), f'__format should be callable, str or number. Was: {__format}'
-
return str(__format(utc_seconds))
-
[docs]
-
def simple_mall_dispatch_core_func(
-
key: KT, action: str, store_name: StoreName, mall: Mall
-
):
-
"""Helper function to dispatch a mall
-
-
This function is only meant to be a helper to give a UI (GUI,
-
CLI...) mall-exploration capabilities. Namely:
-
-
- ``list(mall)``: list the keys of a mall. This is achieved with args:
-
``(key=None, action=None, store_name=None, mall=mall)``
-
- ``mall[store_name]``: get a store. Acheived by:
-
``(key=None, action=None, store_name=store_name, mall=mall)``
-
- ``list(mall[store_name])``: list keys of a store (of the mall). Acheived by:
-
``(key=None, action='list', store_name=store_name, mall=mall)``
-
- ``list(filter(key, mall[store_name]))``: list keys of a store (of the mall)
-
according to a substring filter. (only keys that have ``key`` as substring)
-
``(key=key, action='list', store_name=store_name, mall=mall)``
-
- ``mall[store_name][key]``: get the value/data of a store for ``key``
-
``(key=key, action='get', store_name=store_name, mall=mall)``
-
-
:param key: The key
-
:param action: 'list' (to list keys of a store) or 'get' (to get the value of
-
``key`` in the store (named ``store_name``)
-
:param store_name: Store name to look up in mall. If not given, the function will
-
output the mall keys (which are valid store names)
-
:param mall: dict of stores (Mapping interface to data)
-
:return:
-
-
>>> mall = {
-
... 'english': {'one': 1, 'two': 2, 'three': 3},
-
... 'french': {'un': 1, 'deux': 2},
-
... }
-
-
List the keys of a mall:
-
-
>>> simple_mall_dispatch_core_func(None, None, None, mall=mall)
-
['english', 'french']
-
-
Get a store
-
-
>>> simple_mall_dispatch_core_func(None, None, store_name='english', mall=mall)
-
{'one': 1, 'two': 2, 'three': 3}
-
-
List keys of a store (of the mall):
-
-
>>> simple_mall_dispatch_core_func(
-
... None, action='list', store_name='english', mall=mall
-
... )
-
['one', 'two', 'three']
-
-
List keys of a store (of the mall) according to a substring filter:
-
-
>>> simple_mall_dispatch_core_func(
-
... 'e', action='list', store_name='english', mall=mall
-
... )
-
['one', 'three']
-
-
>>> simple_mall_dispatch_core_func(
-
... 'two', action='get', store_name='english', mall=mall
-
... )
-
2
-
-
-
"""
-
if not store_name:
-
# if store_name empty, list the store names (i.e. the mall keys)
-
return list(mall)
-
else: # if not, get the store
-
store = mall[store_name]
-
if not action:
-
return store
-
-
key = key or ''
-
if action == 'list':
-
key = key.strip() # to handle some invisible whitespace that would screw things
-
return list(filter(lambda k: key in k, store))
-
elif action == 'get':
-
return store[key]
-
[docs]
-
class Crudifier(_Crudifier):
-
"""Convenience class to make crudify (i.e. map/source inputs of) functions.
-
-
See https://github.com/i2mint/front/issues/21.
-
-
``prepare_for_crude_dispatch`` works well if you want to crudify a single function,
-
but if you're trying to crudify multiple functions according to a specific fixed
-
convention, using it directly would involve too much boilerplate.
-
-
``Crudifier`` is one the tools we offer to reduce this boilerplate.
-
-
Here are a few examples of how to use it.
-
-
>>> def foo(x, y):
-
... return x + y
-
...
-
>>> def bar(a, x):
-
... return a * x
-
-
Let's say we want ``x`` to be sourced by the ``x_store`` mapping listed in the
-
``mall``. We can make a ``crudify`` function like this:
-
-
>>> crudify = Crudifier(
-
... param_to_mall_map={'x': 'x_store'}, mall={'x_store': {'stored_two': 2, 'stored_four': 4}}
-
... )
-
-
And apply it to any function containing a argumennt named ``x``:
-
-
>>> from inspect import signature
-
>>> crudified_foo = crudify(foo)
-
>>> str(signature(crudified_foo)) # note how x has now a Literal annotation showing what the valid str inputs are
-
"(x: Literal['stored_two', 'stored_four'], y)"
-
>>> crudified_foo('stored_two', 3) # -> 2 + 3
-
5
-
>>> crudified_bar = crudify(bar)
-
>>> str(signature(crudified_bar))
-
"(a, x: Literal['stored_two', 'stored_four'])"
-
>>> crudified_bar(3, 'stored_two') # -> 3 * 2
-
6
-
-
If the argument names correspond to ``mall`` key, the first ``param_to_mall_map``
-
argument can be specified a list of arguments, or even a space-separated string of
-
these argument names. In the following, the ``'x y'`` is equivalent to
-
``['x', 'y']``, which is equivalent to ``{'x': 'x', 'y', 'y'}``.
-
-
>>> crudify = Crudifier('x y', mall={'x': {'stored_two': 2, 'stored_four': 4}, 'y': {'three': 3}})
-
>>> f = crudify(foo)
-
>>> str(signature(f)) # note that both x and y have a str annotation now
-
"(x: Literal['stored_two', 'stored_four'], y: Literal['three'])"
-
>>> f('stored_two', 'three')
-
5
-
-
This allows you to do things like partialize, to fix the mall, and only have to
-
specify the param_to_mall_map when you want to crudify.
-
In the following, note the ``verbose=False`` which tells the crudification not to
-
issue any warning when it sees we have keys in our ``mall`` that are not arguments
-
of the function.
-
-
>>> from functools import partial
-
>>>
-
>>> mall = {
-
... 'x': {'stored_two': 2}, 'y': {'three': 3}, 'fall_back_store': {'zebra': 11}
-
... }
-
>>> Crudify = partial(Crudifier, mall=mall, verbose=False)
-
>>> f = Crudify('x')(foo)
-
>>> f('stored_two', 3)
-
5
-
>>> f = Crudify('x y')(foo)
-
>>> f('stored_two', 'three')
-
5
-
>>> b = Crudify({'a': 'fall_back_store'})(bar)
-
>>> b('zebra', 3)
-
33
-
-
This callable object, or something like it, can then be used in a recursive
-
transformer such a the front rendering process to indicate that a function should
-
be crudified, and how.
-
-
For example, say we had a mini-language where this
-
-
>>> config = {
-
... foo: {
-
... 'preprocesses': Crudify('x y'),
-
... 'whatevs': 42
-
... },
-
... bar: {
-
... 'blahblah': 24
-
... }
-
... }
-
-
should be preprocessed in such a way that adds a ``'func'`` key to each item of
-
``config`` which contains a transformed function if a ```preprocess`` function
-
or list of functions is specified, or the original function itself otherwise.
-
The following would implement this:
-
-
>>> from typing import Iterable
-
>>> from i2 import Pipe
-
>>>
-
>>> def _ensure_iterable(v):
-
... if not isinstance(v, Iterable):
-
... v = [v]
-
... return v
-
...
-
>>> def prepare(config):
-
... for func, specs in config.items():
-
... if (processes := specs.get('preprocesses', None)) is not None:
-
... preprocess = Pipe(*_ensure_iterable(processes))
-
... _func = preprocess(func)
-
... else:
-
... _func = func
-
... specs = dict(specs, func=_func)
-
... yield func, specs
-
-
>>> prepared_configs = dict(prepare(config))
-
-
Now get the ``func`` value under ``foo``, and see that it has been crudified:
-
-
>>> processed_foo = prepared_configs[foo]['func']
-
>>> processed_foo('stored_two', 'three')
-
5
-
-
"""
-
-
def __call__(self, func):
-
# is there a safer way than vars to get the init fields (keys and values)?
-
return prepare_for_crude_dispatch(func, **vars(self))
-
[docs]
-
def crudify_based_on_names(
-
func, *, param_to_mall_map=(), output_store=(), crudifier=Crudifier
-
):
-
"""
-
Crudify a function based on general
-
-
:param func:
-
:param param_to_mall_map:
-
:param output_store:
-
:param crudifier:
-
:return:
-
-
>>> from functools import partial
-
>>> def foo(x, y):
-
... return x + y
-
>>> def bar(a, x):
-
... return a * x
-
>>> general_crudifier = partial(
-
... crudify_based_on_names,
-
... param_to_mall_map={'x': 'x_store'},
-
... crudifier=partial(prepare_for_crude_dispatch, mall={'x_store': {'stored_two': 2, 'stored_four': 4}})
-
... )
-
>>>
-
>>> foo, bar = map(general_crudifier, [foo, bar])
-
>>>
-
>>> foo('stored_two', 10)
-
12
-
>>> bar(4, 'stored_four')
-
16
-
-
"""
-
param_to_mall_map = dict(param_to_mall_map)
-
output_store = dict(output_store)
-
func_name = name_of_obj(func)
-
param_to_mall_map = (
-
_remove_non_valued_items(
-
{
-
arg_name: chain_get(param_to_mall_map, keys)
-
for arg_name, keys in _keys_to_search(func)
-
}
-
)
-
or None
-
)
-
output_store = chain_get(output_store, (func, func_name), default=None)
-
if param_to_mall_map or output_store:
-
return crudifier(
-
func, param_to_mall_map=param_to_mall_map, output_store=output_store
-
)
-
else:
-
return func
-
[docs]
-
@wrap_kvs(data_of_obj=pickler.dumps, obj_of_data=pickler.loads)
-
class DillFiles(Files):
-
"""Serializes and deserializes with dill (or pickle if dill not installed)"""
-
-
pass
-
[docs]
-
def mk_mall_of_dill_stores(store_names=Iterable[StoreName], rootdir=None):
-
"""Make a mall of DillFiles (or PickleFiles if dill not installed) stores"""
-
rootdir = rootdir or mk_tmp_dol_dir('crude')
-
if isinstance(store_names, str):
-
store_names = store_names.split()
-
-
def name_and_rootdir():
-
for name in store_names:
-
root = os.path.join(rootdir, name)
-
ensure_dir(root)
-
yield name, root
-
-
return {name: DillFiles(root) for name, root in name_and_rootdir()}