Routing: Factories for extendible transformers, mappings, smart defaults etc.

[thorwhalen]

The idea (or solution to a common "problem pattern") of parametrized transformers comes back a lot.

I've denoted the idea by many names such as:

• routing
• parametrized transformers
• declarative transformer
• casting function factories
• wrapping rules
• declarative wrapping
• declarative interpreters

Highlighted here is one of the cornerstone problems. There are many names to this because there's many angles we can approach the problem under. Different contexts bring about different "natural" interfaces to a same underlying problem.

It definitely would be useful to gain a bit more clarity on the underlying ontology so a useful taxonomy can be developed and used to better design reusable tools around this problem. In theory, it's the same underlying idea at the base of "dispatchers" such as py2http and front, and is quite present in other "i2i" tools such as dol.

Below are a few examples of where we actually implemented a tool for some sub-class of the declarative wrapping/transformer solution.

Examples

You get things like the extremely useful (though badly designed, since carries ancient legacies)
StrTupleDict
that is useful to express multi directional transformers from strings, tuples and dictionaries: Used a lot in the context of key transformation when using dol.

The main entropy-reducing part of py2api was it's transformation mini-languages expressed in near-json objects that were then interpreted by interpreters such as InputTrans

• io_trans
• routing_forest
• switch_case_tree

The "Input/Output Trans the i2.wrapper way" mentions that these could probably be boiled down to use i2.wrapper as their backend.

Indeed, i2.wrapper is a pretty solid and general base for such matters, though (1) I wonder if it's a complexity overkill versus having a few specific tools written without so many abstract layers, and (2) I still wonder if we should switch to using wrapt as our base, since it was written by someone who understands the annoying wrap-unfriendly traps deep end python laid out for us all!

Some more mentionables:

• This table getter is already set up declaratively for the "resolve to dataframe" concern, via the ext_mapping, but the io object resolution is parametrized through a resolve_to_io function, with a default that is itself a function whose parameters haven't been factored out to be declarative. In this discussion, we propose that: A declarative form (routing) for the io reader concern
• Conditioning the way you read data according to a file's extension, which was then replaced with this declarative form.
• A declarative generalization of the above to read/write
• A similar routing but applied to url protocols
• Batch crudifying
• Simple function annotator
• Notes on input and output translation

References

• Simple function annotator issue

Quick search links

The links below will take you directly to the github advanced search with a few key terms where you might find more information. These are useful because it will give you a more "current" view of the landscape.

• trans wikis
• trans issues
• routing wikis
• routing issues

┆Issue is synchronized with this Asana task by Unito

[thorwhalen]

Testing 123.

[thorwhalen]

Example of routing: Switch-case clauses.

Switch-case code block

Say you had the following code block in a function.

    if axis_format is None:
        # If the total range is large, we might want to reduce the detail
        if total_range_in_days > 365:  # More than a year
            axis_format = '%Y'
        elif total_range_in_days > 30:  # More than a month
            axis_format = '%Y-%m'
        elif total_range_in_days > 7:  # More than a week
            axis_format = '%Y-%m-%d'
        elif total_range_in_days > 1:  # More than a day
            axis_format = '%Y-%m-%d %H:%M'
        elif total_range_in_days > 0.041666666666666664:  # More than an hour
            axis_format = '%Y-%m-%d %H:%M:%S'
        elif total_range_in_days > 0.0006944444444444444:  # More than a minute
            axis_format = '%Y-%m-%d %H:%M:%S.%L'
        elif total_range_in_days > 0.000011574074074074073:  # More than a second
            axis_format = '%Y-%m-%d %H:%M:%S.%L'
        else:  # Less than a second
            axis_format = '%Y-%m-%d %H:%M:%S.%L'

Well, first of all, since you are a clean-code monk, you'll be tempted to factor out that logic to be in a axis_format_according_to_range function that gets an axis_format in function of a total_range_in_days.

Good. Now if you want your user to be able to control this aspect, you can add an optional parameter to your function with that axis_format_according_to_range as a default.

You're better off now, but what if the context warrants changing that function often, or even worse: Just make some slight modifications to the already existing logic. You are again in danger of copy/paste/modifying a bunch of code, littering your otherwise beautifully concise module. Further, it's not the most convenient thing for the user, and you'll notice that most of the actual logic of the code above is determined by the (total_range_in_days, axis_format) pairs...

We can do better.

IntervalSwitchCase class

Here's an example of better: You have total_range_in_days factory that was easily parametrizable by through the specification of those (total_range_in_days, axis_format) pairs, or one that allowed you to configure what you wanted by only differences with respect to a default/convention.

Here's some starter code:

from collections.abc import Callable

class IntervalSwitchCase(Callable):
    def __init__(self, mapping, default):
        # Store mapping as a sorted tuple
        self.mapping = tuple(sorted(mapping, reverse=True))
        self.default = default

    def __call__(self, value):
        # Implement switch case logic
        for threshold, result in self.mapping:
            if value > threshold:
                return result
        return self.default

    def update(self, updates):
        # Immutably update mapping
        updated_mapping = dict(self.mapping)
        updated_mapping.update(updates)
        return IntervalSwitchCase(list(updated_mapping.items()), self.default)

    def __setitem__(self, key, value):
        # Support item assignment syntax
        updated_mapping = dict(self.mapping)
        updated_mapping[key] = value
        return IntervalSwitchCase(list(updated_mapping.items()), self.default)

    def __delitem__(self, key):
        # Support item deletion syntax
        updated_mapping = dict(self.mapping)
        del updated_mapping[key]
        return IntervalSwitchCase(list(updated_mapping.items()), self.default)

Applying this to our axis_format case, we'd get:

# Define default_mapping and default for axis_format case
axis_format_default_mapping = (
    (365, '%Y'),
    (30, '%Y-%m'),
    (7, '%Y-%m-%d'),
    (1, '%Y-%m-%d %H:%M'),
    (0.041666666666666664, '%Y-%m-%d %H:%M:%S'),
    (0.0006944444444444444, '%Y-%m-%d %H:%M:%S.%L'),
    (1.1574074074074073e-05, '%Y-%m-%d %H:%M:%S.%L')
)
axis_format_default = '%Y-%m-%d %H:%M:%S.%L'

# Create a IntervalSwitchCase instance for the axis_format
axis_format_selector = IntervalSwitchCase(axis_format_default_mapping, axis_format_default)

# Example usage
print(axis_format_selector(50))  # Output: '%Y-%m'

# Get an updated IntervalSwitchCase object with custom mapping
custom_selector = axis_format_selector.update({1: '%Y-%m-%d %H'})
print(custom_selector(0.5))  # Output: '%Y-%m-%d %H'

A general SwitchCase class

Note that this doesn't really implement a general switch case, but what I would call an IntervalSwitchCase where by we determine a value according to where another values lies. In a sense, we're computing axis_format from date_format using a step function.

The general SwitchCase class would take pairs of (callable, value): Tuple[ Callable[..., bool], Any] pairs instead, and call each callable on the input, returning the value if it returns True.

from typing import Callable, Any, Tuple, List


class SwitchCase:
    def __init__(self, mapping: List[Tuple[Callable[..., bool], Any]], default: Any):
        self.mapping = mapping
        self.default = default

    def __call__(self, value):
        # Iterate through each (callable, result) pair
        for predicate, result in self.mapping:
            # If callable returns True, return the corresponding result
            if predicate(value):
                return result
        # If none of the callables return True, return default
        return self.default

    def update(self, updates: List[Tuple[Callable[..., bool], Any]]):
        # Immutably update mapping
        updated_mapping = self.mapping + updates
        return SwitchCase(updated_mapping, self.default)

Our original example could then be replicated as such:

# Define custom callables for each range and corresponding axis format
def more_than_year(value): return value > 365
def more_than_month(value): return value > 30
def more_than_week(value): return value > 7
def more_than_day(value): return value > 1
def more_than_hour(value): return value > 0.041666666666666664
def more_than_minute(value): return value > 0.0006944444444444444
def more_than_second(value): return value > 1.1574074074074073e-05


# Define mapping with custom callables for the axis_format case
axis_format_mapping = [
    (more_than_year, '%Y'),
    (more_than_month, '%Y-%m'),
    (more_than_week, '%Y-%m-%d'),
    (more_than_day, '%Y-%m-%d %H:%M'),
    (more_than_hour, '%Y-%m-%d %H:%M:%S'),
    (more_than_minute, '%Y-%m-%d %H:%M:%S.%L'),
    (more_than_second, '%Y-%m-%d %H:%M:%S.%L')
]
axis_format_default = '%Y-%m-%d %H:%M:%S.%L'

# Create a SwitchCase instance for the axis_format
axis_format_selector = SwitchCase(axis_format_mapping, axis_format_default)

# Example usage
print(axis_format_selector(50))  # Output: '%Y-%m'

# Get an updated SwitchCase object with custom mapping
def more_than_half_day(value): return value > 0.5
custom_selector = axis_format_selector.update([(more_than_half_day, '%Y-%m-%d %H')])
print(custom_selector(0.6))  # Output: '%Y-%m-%d %H'

A factory of interval-switch-case instances

Now, if you're rolling out a lot of "interval switch cases", you may not want to have to write those "is this larger than" functions manually every time. Now you have a general SwitchCase base, you can use it to roll out domain specific factories.

from typing import List, Tuple

def make_interval_switch_case(mapping: List[Tuple[float, Any]], default: Any) -> SwitchCase:
    # Convert interval mapping to callable mapping
    callable_mapping = []
    for interval, result in mapping:
        # Create a callable that checks if the input value is greater than the interval
        callable_mapping.append((lambda interval: lambda value: value > interval)(interval), result)
    
    # Instantiate and return a SwitchCase object with the generated mapping
    return SwitchCase(callable_mapping, default)


# Example usage
interval_mapping = [
    (365, '%Y'),
    (30, '%Y-%m'),
    (7, '%Y-%m-%d'),
    (1, '%Y-%m-%d %H:%M'),
    (0.041666666666666664, '%Y-%m-%d %H:%M:%S'),
    (0.0006944444444444444, '%Y-%m-%d %H:%M:%S.%L'),
    (1.1574074074074073e-05, '%Y-%m-%d %H:%M:%S.%L')
]
default = '%Y-%m-%d %H:%M:%S.%L'

# Create an IntervalSwitchCase instance using the SwitchCase implementation
interval_switch_case = make_interval_switch_case(interval_mapping, default)

# Test the interval_switch_case callable
print(interval_switch_case(50))  # Output: '%Y-%m'

NestedSwitchCase

Finally, we see that SwitchCase is still limited to a linear sequence of checks.

Now, flat is better than nested, in general -- but not always.
So what if you needed a more nested logic.
One that would be the extendible/parametrizable reflection of nested blocks of switch-case clauses?
How would we build on what we have now to reach those nested heights of expression?

from typing import List, Tuple, Callable, Any


class NestedSwitchCase:
    def __init__(self, mapping: List[Tuple[Callable[..., bool], Any]], default: Any):
        self.mapping = mapping
        self.default = default
        
    def __call__(self, *args, **kwargs):
        for condition, result in self.mapping:
            if condition(*args, **kwargs):
                # If result is another instance of NestedSwitchCase, call it
                if isinstance(result, NestedSwitchCase):
                    return result(*args, **kwargs)
                return result
        return self.default

# Example Usage:

# Let's say we want to select an axis format based on total_range_in_days.
# Additionally, if total_range_in_days is more than a year, we want to check another condition
# that uses a different criterion (e.g., is_leap_year).

def is_more_than_a_year(value):
    return value > 365

def is_less_than_a_month(value):
    return value < 30

def is_leap_year(value):
    return value % 4 == 0 and (value % 100 != 0 or value % 400 == 0)

leap_year_mapping = [
    (is_leap_year, "leap year format"),
    (lambda x: True, "non-leap year format")
]

leap_year_switch = NestedSwitchCase(leap_year_mapping, default="non-leap year format")

axis_format_mapping = [
    (is_more_than_a_year, leap_year_switch),
    (is_less_than_a_month, "short format")
]

axis_format_selector = NestedSwitchCase(axis_format_mapping, default="default format")

# Test the nested switch case
print(axis_format_selector(500, 2020))  # Output: "leap year format"
print(axis_format_selector(500, 2021))  # Output: "non-leap year format"
print(axis_format_selector(20, 2020))   # Output: "short format"

[thorwhalen]

Some patterns of routing

[thorwhalen]

object_processing

from meshed import code_to_dag

@code_to_dag
def object_processing():
    processed_obj = process_obj(obj)

object_processing.dot_digraph('rankdir="LR"')

[thorwhalen]

conditional_object_processing

Sometimes the way we want to process an object depends on the object itself.
Of course, we can already have that behavior within the process_obj code, but it's locked inside the code:
The following structure distinguishes the aspect of choosing the processor, and applying it to the obj.

@__import__('meshed').code_to_dag
def conditional_object_processing():
    processor = get_processor_for_obj(obj)
    processed_obj = process_obj(obj, processor)

conditional_object_processing.dot_digraph('rankdir="LR"')

[thorwhalen]

feature_conditional_object_processing

Here, the logic of get_processor_for_obj is subdivided as the composition of

• get_feature: a function that extracts a feature from (i.e. some value representing a characteristic of) the object, and
• get_processor: a function that returns a processor for that feature value

@__import__('meshed').code_to_dag
def feature_conditional_object_processing():
    feature = get_feature(obj)
    processor = get_processor_for_feature(feature)
    processed_obj = process_obj(obj, processor)

feature_conditional_object_processing.dot_digraph('rankdir="LR"')