Last active
June 20, 2022 22:45
-
-
Save hnykda/852e3821da7409e3a1d4f4730b751f6d to your computer and use it in GitHub Desktop.
This file contains hidden or 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
import inspect | |
from typing import Callable, List, Union, Any, Dict | |
Variable = str | |
Value = Union[str, int, float] | |
Condition = List[Value] | |
DataElement = Dict[Any, Any] | |
class BaseCommand: | |
args = None | |
kwargs = None | |
def __init__(self): | |
self.name = self.__class__.__name__ | |
def __call__(self, d: DataElement) -> DataElement: | |
raise NotImplementedError("Every command must implement __call__") | |
def __str__(self) -> str: | |
args_s = "" | |
if self.args: | |
args_s = "".join([f"{x}, " for x in self.args]) | |
if self.kwargs: | |
kwargs_s = "".join([f"{x}={y}, " for x, y in self.kwargs.items()]) | |
return f"{self.name}({args_s}{kwargs_s[:-2]})" | |
else: | |
return f"{self.name}({args_s[:-2]})" | |
def __repr__(self) -> str: | |
return self.__str__() | |
def command_factory(name, op: Callable[..., DataElement]): | |
def __init__(self, *args, **kwargs): | |
self.op = op | |
self.args = args | |
self.kwargs = kwargs | |
BaseCommand.__init__(self) | |
def __call__(self, d: DataElement) -> DataElement: | |
return self.op(d, *self.args, **self.kwargs) | |
doc = "Executes: {}{}\n\n{}".format(op.__name__, inspect.signature(op), op.__doc__) | |
payload = {"__init__": __init__, "__call__": __call__, "__doc__": doc, "op": op} | |
new_class = type(name, (BaseCommand,), payload) | |
return new_class | |
##### | |
import operator | |
# notice we use only three comparators to keep this concise, but | |
# you can easily add `<=` or `>=` | |
operator_mapping = {"<": "lt", ">": "gt", "==": "eq"} | |
def assign( | |
d: DataElement, condition: Condition, target: Variable, value: Value | |
) -> DataElement: | |
var, _op, comp_val = condition | |
op_func = getattr(operator, operator_mapping[_op]) | |
if op_func(d[var], comp_val): | |
d[target] = value | |
return d | |
def delete(d: DataElement, target: Variable) -> DataElement: | |
del d[target] | |
return d | |
Assign = command_factory("Assign", assign) | |
Delete = command_factory("Delete", delete) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment