mapper.py

from __future__ import annotations

from collections import deque
from datetime import date
from pprint import pprint
from typing import Any, Callable, Dict, List, Type, Union

import attr


def is_int(value: str) -> bool:
    try:
        int(value)
        return True
    except ValueError:
        return False


def int_or_str(value: str) -> Union[int, str]:
    try:
        return int(value)
    except ValueError:
        return value


def transform(src: Dict[str, str], maps: List[Map]):
    res: Dict = {}

    for mapping in maps:
        set_value(mapping, res)

    return res


def set_value(mapping: Map, res: dict):
    value = mapping.get_value(src)
    dst_list = deque(mapping.get_dst_list())
    set_path_value(res, dst_list, value)


def set_path_value(res, path, value):
    if not path:
        # at the leaf
        return

    # current index as a int or str
    current = int_or_str(path.popleft())

    if path:
        # we are not at the leaf yet
        # try to find the next data type (list or dict)
        # next default type is a dict
        next_type = {}
        if is_int(path[0]):
            # next type is a list
            next_type = []
        try:
            # already initialize, do nothing
            res[current]
        except KeyError:
            # does not exists, initialize from a dict
            res[current] = next_type
        except IndexError:
            # does not exists, initialize from a list
            res.append(next_type)
    elif not path:
        # path is now empty, set the value in a dict or list
        try:
            res[current] = value
        except IndexError:
            res.append(value)

    set_path_value(res[current], path, value)


@attr.s(auto_attribs=True)
class Map:
    origin: Union[str, List[str]]
    destination: str
    dst_type: Callable = str

    def get_value(self, src: Dict[str, str]) -> str:

        if isinstance(self.origin, list):
            type_args = [src[key] for key in self.origin]
        else:
            type_args = [src[self.origin]]

        return self.dst_type(*type_args)

    def get_dst_list(self) -> List[str]:
        return self.destination.split(".")


def to_date(year, month, day):
    return date(int(year), int(month), int(day))


maps = [
    Map("key1", "Destination.attrOne", int),
    Map("key2", "Destination.attr2"),
    Map("key3", "Destination.attr3.0"),
    Map("key4", "Destination.attr3.1"),
    Map("key5", "Destination.list.0.key5"),
    Map("key6", "Destination.list.0.key6"),
    Map("key7", "Destination.list.1.key7"),
    Map("key8", "Destination.list.1.key8"),
    Map(["key9", "key10", "key11"], "Destination.date", to_date),
]


src = {
    "key1": "10",
    "key2": "key2",
    "key3": "key3",
    "key4": "key4",
    "key5": "key5",
    "key6": "key6",
    "key7": "key7",
    "key8": "key8",
    "key9": "2019",
    "key10": "01",
    "key11": "01",
}
res = transform(src, maps)

pprint(maps)
pprint(res)

output

$ pipenv run python mapper.py 
[Map(origin='key1', destination='Destination.attrOne', dst_type=<class 'int'>),
 Map(origin='key2', destination='Destination.attr2', dst_type=<class 'str'>),
 Map(origin='key3', destination='Destination.attr3.0', dst_type=<class 'str'>),
 Map(origin='key4', destination='Destination.attr3.1', dst_type=<class 'str'>),
 Map(origin='key5', destination='Destination.list.0.key5', dst_type=<class 'str'>),
 Map(origin='key6', destination='Destination.list.0.key6', dst_type=<class 'str'>),
 Map(origin='key7', destination='Destination.list.1.key7', dst_type=<class 'str'>),
 Map(origin='key8', destination='Destination.list.1.key8', dst_type=<class 'str'>),
 Map(origin=['key9', 'key10', 'key11'], destination='Destination.date', dst_type=<function to_date at 0x7ff076a37e18>)]
{'Destination': {'attr2': 'key2',
                 'attr3': ['key3', 'key4'],
                 'attrOne': 10,
                 'date': datetime.date(2019, 1, 1),
                 'list': [{'key5': 'key5', 'key6': 'key6'},
                          {'key7': 'key7', 'key8': 'key8'}]}}

Pipfile

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]

[packages]
attrs = "*"
isort = "*"
black = "==19.3b0"

[requires]
python_version = "3.7"