A declarative form (routing) for the io reader concern

[thorwhalen]

See also:

• conversation with chatGPT
• routing discussion
• ... and many more musings and code about this throughout i2mint (and thorwhalen) github discussions, issues, and code.

Again the theme of "routing" when refactoring code towards reuse (declarative form, plugin architecture, framework).

In tabled.base I 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.

The current set up is already quite "frameworky", as you can see:

from typing import BinaryIO

TableSrc = Union[str, bytes, BinaryIO]
BinaryIOCaster = Callable[[TableSrc], BinaryIO]

# TODO: Routing pattern!
def default_io_resolver(src: TableSrc) -> BinaryIO:
    if isinstance(src, str):
        if os.path.isfile(src):
            return open(src, 'rb')
        elif get_protocol(src) in {'http', 'https'}:
            import requests

            return BytesIO(requests.get(src).content)
        else:
            raise ValueError(f"Can't handle source: {src}")
    elif isinstance(src, bytes):
        return BytesIO(src)
    elif hasattr(src, 'read'):
        return src
    else:
        raise ValueError(f"Can't handle source: {src}")


def get_table(
    table_src: TableSrc = None,
    *,
    ext=None,
    mapping=dflt_ext_mapping,
    resolve_to_io=default_io_resolver,
    **extra_decoder_kwargs,
):
    """
    Get a table from a variety of sources.
    """
    # If table_src is None, the user is trying to fix the parameters of the function
    if table_src is None:
        return partial(
            ext=ext, mapping=mapping, get_io_obj=get_io_obj, **extra_decoder_kwargs
        )
    
    # Get a BinaryIO object from the source
    io_reader = resolve_to_io(table_src)
    
    return df_from_data_given_ext(
        io_reader, ext=ext, mapping=mapping, **extra_decoder_kwargs
    )

But when I'm at this point of my refactoring towards reuse (smaller functions, etc.) I'm tempted to go a step of "frameworking" or "plugin architecture" further and note that the default_io_resolver is really parametrized by a nested mapping of conditions and resulting returned functions.

I could enable to user to parametrize the io resolution in a "declarative" manner, that is, by editing the "configuration" that the nested mapping of conditions is, instead of having to write their own function.

For example, my resolve_io (corresponding to the above default_io_resolver function could be like this:

def _isinstance(obj, class_or_tuple):
    return isinstance(obj, class_or_tuple)

def instanceof(class_or_tuple):
    return partial(_isinstance, class_or_tuple=class_or_tuple)

def raise_error(error):
    def _raise_error(*args, **kwargs):
        raise error
    raise _raise_error

def always_true(x):
    return True

def identity(x):
    return x    

def is_reader(x):
    return hasattr(x, 'read')

ELSE = always_true


import os
from io import BytesIO
import requests

from tabled.base import get_protocol


def get_content_bytesio(url):
    import requests  # pip install requests

    return BytesIO(requests.get(url).content)


io_resolver_tree = {
    instanceof(str): {
        os.path.isfile: partial(open, mode='rb'),
        lambda s: get_protocol(s) in {'http', 'https'}: get_content_bytesio,
        ELSE: lambda src: raise_error(ValueError(f"Can't handle source: {src}"))
    },
    instanceof(bytes): BytesIO,
    is_reader: lambda x: x,
    ELSE: lambda src: raise_error(ValueError(f"Can't handle source: {src}"))
}

def resolve_io(src, resolver_tree=io_resolver_tree, *, recursive_type=dict):
    for condition, resolver in resolver_tree.items():
        if condition(src):
            if isinstance(resolver, recursive_type):
                return resolve_io(src, resolver_tree=resolver, recursive_type=recursive_type)
            return resolver(src)
    raise ValueError(f"Can't handle source: {src}")


def test_resolve_io():
    def bytes_are_as_expected(bytes_):
        return isinstance(bytes_, bytes) and bytes_ == b'<!doctype '

    # test on url
    io_obj_from_url = resolve_io('https://www.google.com')
    bytes_ = io_obj_from_url.read(10)
    assert bytes_are_as_expected(bytes_), f"io_obj_from_url led to unexpected bytes: {bytes_}"

    # test on filepath and file pointer

    import tempfile
    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(bytes_)
        f.close()
        temp_path = f.name
        io_obj_from_file = resolve_io(temp_path)
        bytes_ = io_obj_from_file.read(10)
        assert bytes_are_as_expected(bytes_), f"io_obj_from_file led to unexpected bytes: {bytes_}"

        with open(temp_path, 'rb') as f:
            io_obj_from_file_pointer = resolve_io(f)
            bytes_ = io_obj_from_file_pointer.read(10)
            assert bytes_are_as_expected(bytes_), f"io_obj_from_file_pointer led to unexpected bytes: {bytes_}"



    # test on bytes
    io_obj_from_bytes = resolve_io(bytes_)
    bytes_ = io_obj_from_bytes.read(10)
    assert bytes_are_as_expected(bytes_), f"io_obj_from_bytes led to unexpected bytes: {bytes_}"

    # test on reader
    io_obj_from_reader = resolve_io(BytesIO(bytes_))
    bytes_ = io_obj_from_reader.read(10)
    assert bytes_are_as_expected(bytes_), f"io_obj_from_reader led to unexpected bytes: {bytes_}"

But when I do that, I then wonder if I've pushed design too far again, prematurely.
The design is better, I have little doubt about that.
But even better design comes at a cost, not even counting the cost of actually designing the thing: Complexity and readability-to-the-novice.