Source code for caspia.reactive.operations.operators

from ..observable import Observable
from ..utils import extension_method
from ..value import Value
from .lambda_ import lambda_


@extension_method(Observable)
def _make_binary_operation(self, other, operator, description):
    if not isinstance(other, Observable):
        other = Value(other)
    return lambda_(operator, self, other, desc=description)


def _define_binary_operator(function, description):
    def operator(self, other):
        # pylint: disable=protected-access
        return self._make_binary_operation(other, function, description)

    return operator


operators = [
    ('__add__', lambda a, b: a + b, '{} + {}'),
    ('__sub__', lambda a, b: a - b, '{} - {}'),
    ('__mul__', lambda a, b: a * b, '{} * {}'),
    ('__truediv__', lambda a, b: a / b, '{} / {}'),
    ('__floordiv__', lambda a, b: a // b, '{} // {}'),
    ('__mod__', lambda a, b: a % b, '{} % {}'),
    ('__and__', lambda a, b: a and b, '{} and {}'),
    ('__or__', lambda a, b: a or b, '{} or {}'),
    ('__lt__', lambda a, b: a < b, '{} < {}'),
    ('__le__', lambda a, b: a <= b, '{} <= {}'),
    ('__eq__', lambda a, b: a == b, '{} == {}'),
    ('__ne__', lambda a, b: a != b, '{} != {}'),
    ('__gt__', lambda a, b: a > b, '{} > {}'),
    ('__ge__', lambda a, b: a >= b, '{} >= {}'),
    ('__radd__', lambda b, a: a + b, '{} + {}'),
    ('__rsub__', lambda b, a: a - b, '{} - {}'),
    ('__rmul__', lambda b, a: a * b, '{} * {}'),
    ('__rtruediv__', lambda b, a: a / b, '{} / {}'),
    ('__rfloordiv__', lambda b, a: a // b, '{} // {}'),
    ('__rmod__', lambda b, a: a % b, '{} % {}'),
    ('__rand__', lambda b, a: a and b, '{} and {}'),
    ('__ror__', lambda b, a: a or b, '{} or {}'),
]

for name, func, desc in operators:
    extension_method(Observable, name)(_define_binary_operator(func, desc))