"""
Utilities for caching price info
"""
from decimal import Decimal
from shuup.core.models import AnonymousContact, ShopProduct
from shuup.core.pricing import PriceInfo
from shuup.core.utils import context_cache
from shuup.utils.dates import to_timestamp
PRICE_INFO_NAMESPACE_ITEM = "price_info_%(shop_id)s"
def _get_price_info_namespace_for_shop(shop_id):
return PRICE_INFO_NAMESPACE_ITEM % {"shop_id": shop_id}
def _get_price_info_cache_key_params(context, item, quantity, **context_args):
shop_id = context.shop.pk if hasattr(context, "shop") else 0
customer = getattr(context, "customer", None)
if customer:
cached_customer_groups = getattr(customer, "_cached_customer_groups", None)
if cached_customer_groups:
groups = customer._cached_customer_groups
else:
groups = list(customer.groups.order_by("pk").values_list("pk", flat=True))
customer._cached_customer_groups = groups
else:
anonymous_group_id = getattr(AnonymousContact, "_cached_default_group_id", None)
if anonymous_group_id:
groups = [AnonymousContact._cached_default_group_id]
else:
anonymous_group_id = AnonymousContact().default_group.pk
AnonymousContact._cached_default_group_id = anonymous_group_id
groups = [anonymous_group_id]
extra_kwargs = {}
for key, value in context_args.items():
if hasattr(value, "pk"):
extra_kwargs[key] = value.pk
else:
extra_kwargs[key] = value
return dict(
identifier="price_info_cache",
item=_get_price_info_namespace_for_shop(shop_id),
context={},
customer_groups=groups,
quantity=str(Decimal(quantity)),
item_id=item.pk if hasattr(item, "pk") else str(item),
**extra_kwargs,
)
[docs]
def cache_many_price_info(context, item, quantity, prices_infos, **context_args):
"""
Cache a list of PriceInfo
:param object|WSGIRequest context: the context should contain at least a shop and a customer property
:param object item
:param float|Decimal quantity
:param iterable[PriceInfo] prices_infos
"""
# check whether the prices are iterable
try:
iter(prices_infos)
except TypeError:
return
# all items must be PriceInfo
if not all(isinstance(item, PriceInfo) for item in prices_infos):
return
context_kwargs = _get_price_info_cache_key_params(context, item, quantity, **context_args)
key = context_cache.get_cache_key_for_context(many=True, **context_kwargs)
context_cache.set_cached_value(key, prices_infos)
[docs]
def cache_price_info(context, item, quantity, price_info, **context_args):
"""
Cache a PriceInfo
:param context object|WSGIRequest: the context should contain at least a shop and a customer property
:param item any
:param quantity float|Decimal
:param price_info PriceInfo
"""
# we can just cache PriceInfo instances
if isinstance(price_info, PriceInfo):
key = context_cache.get_cache_key_for_context(
**_get_price_info_cache_key_params(context, item, quantity, **context_args)
)
context_cache.set_cached_value(key, price_info)
[docs]
def get_many_cached_price_info(context, item, quantity=1, **context_args):
"""
Get cached prices info list
:param object|WSGIRequest context: the context should contain at least a shop and a customer property
:param object item
:param float|Decimal quantity
"""
cache_kwargs = _get_price_info_cache_key_params(context, item, quantity, many=True, **context_args)
key, prices_infos = context_cache.get_cached_value(**cache_kwargs)
if prices_infos:
try:
iter(prices_infos)
except TypeError:
return None
from django.utils.timezone import now
now_timestamp = to_timestamp(now())
# make sure to check all experiration dates
for price_info in prices_infos:
# if one price has expired, we invalidate the entire cache
if isinstance(price_info, PriceInfo) and price_info.expires_on and price_info.expires_on < now_timestamp:
return None
return prices_infos
[docs]
def get_cached_price_info(context, item, quantity=1, **context_args):
"""
Get a cached price info
:param object|WSGIRequest context: the context should contain at least a shop and a customer property
:param object item
:param float|Decimal quantity
"""
cache_kwargs = _get_price_info_cache_key_params(context, item, quantity, **context_args)
key, price_info = context_cache.get_cached_value(**cache_kwargs)
from django.utils.timezone import now
now_ts = to_timestamp(now())
# price has expired
if price_info and isinstance(price_info, PriceInfo) and price_info.expires_on and price_info.expires_on < now_ts:
price_info = None
return price_info
[docs]
def bump_price_info_cache(shop):
"""
Bump the price info cache for the entire shop
:param int|Shop shop: the shop to be bump caches
"""
from shuup.core.models import Shop
shop_id = shop.pk if isinstance(shop, Shop) else int(shop)
context_cache.bump_cache_for_item(_get_price_info_namespace_for_shop(shop_id))
[docs]
def bump_all_price_caches(shops=None):
"""
Bump all price info caches for the given shops or all shops
:param list[int|Shop]|None shops: the shops list to be bump caches or None to bump all
"""
from shuup.core.models import Shop
if shops is None:
shops = []
if shops:
shop_ids = [shop.pk if isinstance(shop, Shop) else int(shop) for shop in shops]
else:
shop_ids = Shop.objects.values_list("pk", flat=True)
for shop_id in shop_ids:
bump_price_info_cache(shop_id)
[docs]
def bump_prices_for_product(product_id):
if hasattr(product_id, "pk"):
product_id = product_id.pk
for shop_id in set(ShopProduct.objects.filter(product_id=product_id).values_list("shop_id", flat=True)):
bump_price_info_cache(shop_id)
[docs]
def bump_prices_for_shop_product(shop_id):
if isinstance(shop_id, ShopProduct):
shop_id = shop_id.shop_id
bump_price_info_cache(shop_id)