from decimal import Decimal
from functools import lru_cache
from typing import TYPE_CHECKING, Dict, Iterable, Optional
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from shuup.apps.provides import get_provide_objects
from shuup.core.specs.product_kind import ProductKindSpec
from shuup.core.stocks import ProductStockStatus
from .enums import StockAdjustmentType
USER_MODEL = get_user_model()
if TYPE_CHECKING: # pragma: no cover
from shuup.core.models import Contact, Product, Shipment, ShopProduct, Supplier
[docs]
@lru_cache
def get_supported_product_kinds_for_module(
module_identifier: str,
) -> Iterable[ProductKindSpec]:
specs = []
for product_kind_spec in get_provide_objects("product_kind_specs"):
supported_modules = product_kind_spec.supported_supplier_modules
if not supported_modules or module_identifier in supported_modules:
specs.append(product_kind_spec)
return specs
[docs]
@lru_cache
def get_supported_product_kinds_values_for_module(
module_identifier: str,
) -> Iterable[int]:
return [spec.value for spec in get_supported_product_kinds_for_module(module_identifier)]
[docs]
class SupplierModuleInterface:
"""
Supplier module interface.
"""
identifier = None # type: str
name = None # type: str
[docs]
def get_stock_statuses(self, product_ids, *args, **kwargs) -> Dict[int, ProductStockStatus]:
"""
:param product_ids: Iterable of product IDs.
:return: Dict of {product_id: ProductStockStatus}.
"""
return {}
[docs]
def get_stock_status(self, product_id: int, *args, **kwargs) -> Optional[ProductStockStatus]:
"""
:param product_id: Product ID.
:type product_id: int
:rtype: shuup.core.stocks.ProductStockStatus|None
"""
return None
[docs]
def get_orderability_errors(
self, shop_product: "ShopProduct", quantity: Decimal, customer: "Contact"
) -> Iterable[ValidationError]:
"""
:param shop_product: Shop Product.
:type shop_product: shuup.core.models.ShopProduct
:param quantity: Quantity to order.
:type quantity: decimal.Decimal
:param customer: Contact.
"""
return []
[docs]
def adjust_stock(
self,
product_id: int,
delta: Decimal,
created_by: USER_MODEL = None,
type: StockAdjustmentType = StockAdjustmentType.INVENTORY,
*args,
**kwargs,
) -> None:
"""
Adjusts the stock for the given `product_id`.
"""
pass
[docs]
def update_stock(self, product_id: int, *args, **kwargs) -> None:
"""
Updates a stock for the given `product_id`
"""
pass
[docs]
def update_stocks(self, product_ids: Iterable[int], *args, **kwargs) -> None:
pass
[docs]
def ship_products(
self,
shipment: "Shipment",
product_quantities: Dict["Product", Decimal],
*args,
**kwargs,
):
pass
[docs]
@classmethod
def get_supported_product_kinds(cls) -> Iterable[ProductKindSpec]:
raise NotImplementedError
[docs]
@classmethod
def get_supported_product_kinds_values(cls) -> Iterable[int]:
raise NotImplementedError
[docs]
class BaseSupplierModule(SupplierModuleInterface):
"""
Base supplier module implementation.
"""
[docs]
def __init__(self, supplier: "Supplier", options: Dict):
"""
:type supplier: Supplier.
:type options: dict
"""
self.supplier = supplier
self.options = options
[docs]
@classmethod
def get_supported_product_kinds(cls) -> Iterable[ProductKindSpec]:
return get_supported_product_kinds_for_module(cls.identifier)
[docs]
@classmethod
def get_supported_product_kinds_values(cls) -> Iterable[int]:
return get_supported_product_kinds_values_for_module(cls.identifier)
[docs]
def get_stock_status(self, product_id: int, *args, **kwargs):
statuses = self.get_stock_statuses([product_id])
if product_id in statuses:
return statuses[product_id]
[docs]
def update_stocks(self, product_ids: Iterable[int], *args, **kwargs):
# Naive default implementation; smarter modules can do something better
for product_id in product_ids:
self.update_stock(product_id)