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_request.py
Size: Mime:
from typing import TypeVar, Type
from lib_b2b.change import ChangeRequest
from enum import Enum

from lib_b2b.request import Request


class OrderRequestType(Enum):
    CANCEL = "orders/cancelled"
    CREATE = "orders/create"
    FULFILLED = "orders/fulfilled"
    PAID = "orders/paid"
    PARTIALLY_FULFILLED = "orders/partially_fulfilled"
    CHANGED = "orders/updated"


class OrderRequest:
    C = TypeVar('C', bound=ChangeRequest)

    @staticmethod
    def should_suppress_http_errors_for(request):
        if 'x-shopify-shop-domain' in request.headers or 'X-Shopify-Shop-Domain' in request.headers:
            return True
        return False

    @staticmethod
    def for_(request: Request):
        if 'x-shopify-shop-domain' in request.headers or 'X-Shopify-Shop-Domain' in request.headers:
            return ShopifyOrderRequest(request)
        else:
            return OrderRequest(request)

    def __init__(self, request: Request):
        self.request = request
        from .dialect import Dialect
        self.dialect = Dialect.infer_from(request)
        request_type = request.request_type or self.dialect.request_type
        self.dialect.verify(request_type)
        self.request_data = self.request.json_body
        self.headers = request.headers
        self.request_type = request_type

    @property
    def order_id(self):
        return self.dialect.order_id

    @property
    def principle_id(self):
        # TODO: - Test with the deployed app to make sure I'm grabbing the right context variable
        try:
            if self.request.context:
                return self.request.context.get('authorizer').get('principalId')
            else:
                from lib_b2b.profile import Profile
                profile = Profile.profile_for(self.dialect.customer)
                return profile.sender_address.company
        except AttributeError:
            return "System"

    def data_provider_for(self, order_request: 'OrderRequest', change_request_type: Type[C]):
        return self.dialect.data_provider_for(self, order_request, change_request_type)

    @property
    def channel_order_id(self):
        return self.purchase_order

    def __getattr__(self, attr):
        try:
            obj = self.request_data[attr]
        except (TypeError, KeyError):
            obj = getattr(self.request_data, attr, None)
        return obj


class ShopifyOrderRequest(OrderRequest):
    def __init__(self, request):
        super().__init__(request)

    @property
    def shopify_domain(self):
        return self.headers.get('x-shopify-shop-domain', self.headers.get('X-Shopify-Shop-Domain', None))

    @property
    def topic(self):
        return self.headers.get('x-shopify-topic', None)

    @property
    def channel_order_id(self):
        return self.headers.get('x-shopify-order-id', None)

    @property
    def signature(self):
        return self.headers.get('x-shopify-hmac-sha256', None)