from abc import abstractmethod, ABC
import typing as tp
from itertools import groupby
from operator import itemgetter

TRow = tp.Dict[str, tp.Any]
TRowsIterable = tp.Iterable[TRow]
TRowsGenerator = tp.Generator[TRow, None, None]


class Operation(ABC):
    @abstractmethod
    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        pass


# Operations


class Mapper(ABC):
    """Base class for mappers"""

    @abstractmethod
    def __call__(self, row: TRow) -> TRowsGenerator:
        """
        :param row: one table row
        """
        pass


class Map(Operation):
    def __init__(self, mapper: Mapper) -> None:
        self.mapper = mapper

    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        for row in rows:
            yield from self.mapper(row)


class Reducer(ABC):
    """Base class for reducers"""

    @abstractmethod
    def __call__(self, group_key: tp.Tuple[str, ...], rows: TRowsIterable) -> TRowsGenerator:
        """
        :param rows: table rows
        """

        pass


class Reduce(Operation):
    def __init__(self, reducer: Reducer, keys: tp.Sequence[str]) -> None:
        self.reducer = reducer
        self.keys = keys

    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        if self.keys:
            for group_keys, group_rows in groupby(rows, key=itemgetter(*self.keys)):
                yield from self.reducer(tuple(self.keys), group_rows)
        else:
            yield from self.reducer(tuple(self.keys), rows)


class Joiner(ABC):
    """Base class for joiners"""

    def __init__(self, suffix_a: str = '_1', suffix_b: str = '_2') -> None:
        self._a_suffix = suffix_a
        self._b_suffix = suffix_b

    @abstractmethod
    def __call__(self, keys: tp.Sequence[str], rows_a: TRowsIterable, rows_b: TRowsIterable) -> TRowsGenerator:
        """
        :param keys: join keys
        :param rows_a: left table rows
        :param rows_b: right table rows
        """
        pass

    def update(self, dict_a: TRow, dict_b: TRow, keys: tp.Sequence[str]) -> None:
        for key, val_b in dict_b.items():
            if key in dict_a and key not in keys:
                val_a = dict_a[key]
                del dict_a[key]
                dict_a['{}{}'.format(key, self._a_suffix)] = val_a
                dict_a['{}{}'.format(key, self._b_suffix)] = val_b
            else:
                dict_a[key] = val_b

from abc import abstractmethod, ABC
import typing as tp
from itertools import groupby
from operator import itemgetter

TRow = tp.Dict[str, tp.Any]
TRowsIterable = tp.Iterable[TRow]
TRowsGenerator = tp.Generator[TRow, None, None]


class Operation(ABC):
    @abstractmethod
    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        pass


# Operations


class Mapper(ABC):
    """Base class for mappers"""

    @abstractmethod
    def __call__(self, row: TRow) -> TRowsGenerator:
        """
        :param row: one table row
        """
        pass


class Map(Operation):
    def __init__(self, mapper: Mapper) -> None:
        self.mapper = mapper

    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        for row in rows:
            yield from self.mapper(row)


class Reducer(ABC):
    """Base class for reducers"""

    @abstractmethod
    def __call__(self, group_key: tp.Tuple[str, ...], rows: TRowsIterable) -> TRowsGenerator:
        """
        :param rows: table rows
        """

        pass


class Reduce(Operation):
    def __init__(self, reducer: Reducer, keys: tp.Sequence[str]) -> None:
        self.reducer = reducer
        self.keys = keys

    def __call__(self, rows: TRowsIterable, *args: tp.Any, **kwargs: tp.Any) -> TRowsGenerator:
        if self.keys:
            for group_keys, group_rows in groupby(rows, key=itemgetter(*self.keys)):
                yield from self.reducer(tuple(self.keys), group_rows)
        else:
            yield from self.reducer(tuple(self.keys), rows)


class Joiner(ABC):
    """Base class for joiners"""

    def __init__(self, suffix_a: str = '_1', suffix_b: str = '_2') -> None:
        self._a_suffix = suffix_a
        self._b_suffix = suffix_b

    @abstractmethod
    def __call__(self, keys: tp.Sequence[str], rows_a: TRowsIterable, rows_b: TRowsIterable) -> TRowsGenerator:
        """
        :param keys: join keys
        :param rows_a: left table rows
        :param rows_b: right table rows
        """
        pass

    def update(self, dict_a: TRow, dict_b: TRow, keys: tp.Sequence[str]) -> None:
        for key, val_b in dict_b.items():
            if key in dict_a and key not in keys:
                val_a = dict_a[key]
                del dict_a[key]
                dict_a['{}{}'.format(key, self._a_suffix)] = val_a
                dict_a['{}{}'.format(key, self._b_suffix)] = val_b
            else:
                dict_a[key] = val_b

# Operations Base class for mappers :param row: one table row Base class for reducers :param rows: table rows Base class for joiners :param keys: join keys         :param rows_a: left table rows         :param rows_b: right table rows