Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
lib-py-b2b / order_change / __init__.py
Size: Mime:
from lib_b2b.policy import Predicate, SimplePredicate
from typing import Callable
from lib_b2b.change import ChangeRequest, ChangeDataProvider, Changeable
import logging
from os import environ
import abc

log_level = logging.getLevelName(environ['LOG_LEVEL']) if 'LOG_LEVEL' in environ else logging.INFO
logging.basicConfig(format='%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s',
                    datefmt='%d-%m-%Y:%H:%M:%S',
                    level=log_level)
logger = logging.getLogger('lib-b2b-order-change')


class OrderChangeRequest(ChangeRequest):

    def __init__(self, on_behalf_of: str, conditions: [Predicate]):
        super().__init__(on_behalf_of, conditions)

    @abc.abstractmethod
    def will_change(self, order: 'Order') -> bool:
        """
        Compares the existing data to the change request to determine if there has been a change
        :return: True/False
        :rtype: bool
        """
        raise NotImplementedError

    def permitted(self, changeable_object: 'Changeable') -> bool:
        self.changeable_object = changeable_object
        return self.policy.is_permitted(self, changeable_object)

    def operation_identifier(self) -> str:
        from lib_b2b.order import Order
        if isinstance(self.changeable_object, Order):
            return f"{type(self).__name__} on {self.changeable_object.order_id}"
        else:
            return f"{type(self).__name__} on {str(self.changeable_object)}"

    def subject(self) -> str:
        from lib_b2b.order import Order
        if isinstance(self.changeable_object, Order):
            return self.changeable_object.customer_edi_id
        else:
            return str(self.changeable_object)

    @abc.abstractmethod
    def apply(self, order: 'Order'):
        raise NotImplementedError


class SimpleOrderPredicate(SimplePredicate):

    def __init__(self, fn: Callable[[Changeable, ChangeRequest], bool], name: str):
        super().__init__(fn, name)

    def evaluate(self, changeable: Changeable, with_: ChangeRequest) -> bool:
        result = self.fn(changeable)
        return result


class OrderStatusChangeDataProvider(ChangeDataProvider, metaclass=abc.ABCMeta):
    @property
    @abc.abstractmethod
    def new_status(self) -> 'OrderStatus':
        raise NotImplementedError

from . import accept
from . import address
from . import cancel
from . import cancel_line
from . import paid
from . import ship_date
from . import revision