Egor fureunoir Gorbunov
0db5b9b712
Add tailored querysets and managers for `Product`, `Stock`, and `AttributeValue` models to simplify vendor-specific operations. Abstract methods were added to `AbstractVendor` for handling essential vendor processes, ensuring clear API expectations. This change standardizes vendor-related queries, improves clarity, and ensures efficient bulk operations.
246 lines
9 KiB
Python
246 lines
9 KiB
Python
import logging
|
|
from typing import TYPE_CHECKING, Self
|
|
|
|
import requests
|
|
from constance import config
|
|
from django.contrib.gis.geos import Point
|
|
from django.db import models
|
|
from django.db.models import QuerySet
|
|
from modeltranslation.manager import MultilingualManager, MultilingualQuerySet
|
|
|
|
if TYPE_CHECKING:
|
|
from engine.core.models import AttributeValue, Product, Stock, Vendor # noqa: F401
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AddressManager(models.Manager):
|
|
def create(self, **kwargs):
|
|
if not kwargs.get("raw_data"):
|
|
raise ValueError("'raw_data' (address string) must be provided.")
|
|
|
|
params: dict[str, str | int | None] = {
|
|
"format": "json",
|
|
"addressdetails": 1,
|
|
"q": kwargs.get("raw_data"),
|
|
}
|
|
resp = requests.get(config.NOMINATIM_URL.rstrip("/") + "/search", params=params)
|
|
resp.raise_for_status()
|
|
results = resp.json()
|
|
if not results:
|
|
raise ValueError(
|
|
f"No geocoding result for address: {kwargs.get('raw_data')}"
|
|
)
|
|
data = results[0]
|
|
|
|
addr = data.get("address", {})
|
|
street = f"{addr.get('road', '') or addr.get('pedestrian', '')}, {addr.get('house_number', '')}"
|
|
district = addr.get("city_district") or addr.get("suburb") or ""
|
|
city = addr.get("city") or addr.get("town") or addr.get("village") or ""
|
|
region = addr.get("state") or addr.get("region") or ""
|
|
postal_code = addr.get("postcode") or ""
|
|
country = addr.get("country") or ""
|
|
|
|
try:
|
|
lat = float(data.get("lat"))
|
|
lon = float(data.get("lon"))
|
|
location = Point(lon, lat, srid=4326)
|
|
except (TypeError, ValueError):
|
|
location = None
|
|
|
|
try:
|
|
address_line_1 = kwargs.pop("address_line_1")
|
|
except KeyError as e:
|
|
raise ValueError("Missing required field 'address_line_1'") from e
|
|
|
|
try:
|
|
address_line_2 = kwargs.pop("address_line_2")
|
|
except KeyError:
|
|
address_line_2 = ""
|
|
|
|
return super().get_or_create(
|
|
raw_data=kwargs.get("raw_data"),
|
|
address_line=f"{address_line_1}, {address_line_2}",
|
|
street=street,
|
|
district=district,
|
|
city=city,
|
|
region=region,
|
|
postal_code=postal_code,
|
|
country=country,
|
|
user=kwargs.pop("user"),
|
|
defaults={"api_response": data, "location": location},
|
|
)[0]
|
|
|
|
|
|
class ProductQuerySet(MultilingualQuerySet["Product"]):
|
|
"""Custom QuerySet for Product with vendor-related operations."""
|
|
|
|
def available(self) -> Self:
|
|
"""Filter to only available products."""
|
|
return self.filter(
|
|
is_active=True,
|
|
brand__is_active=True,
|
|
category__is_active=True,
|
|
stocks__isnull=False,
|
|
stocks__vendor__is_active=True,
|
|
)
|
|
|
|
def available_in_stock(self) -> Self:
|
|
"""Filter to available products with stock quantity > 0."""
|
|
return self.filter(
|
|
is_active=True,
|
|
brand__is_active=True,
|
|
category__is_active=True,
|
|
stocks__isnull=False,
|
|
stocks__vendor__is_active=True,
|
|
stocks__quantity__gt=0,
|
|
)
|
|
|
|
def with_related(self) -> Self:
|
|
"""Prefetch related objects for performance."""
|
|
return self.select_related("category", "brand").prefetch_related(
|
|
"tags", "stocks", "images", "attributes__attribute__group"
|
|
)
|
|
|
|
def stale(self) -> Self:
|
|
"""Filter products with no orders and no stocks."""
|
|
return self.filter(orderproduct__isnull=True, stocks__isnull=True)
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> Self:
|
|
"""Filter products that have stocks from a specific vendor."""
|
|
return self.filter(stocks__vendor=vendor)
|
|
|
|
def not_in_orders(self) -> Self:
|
|
"""Filter products that are not part of any order."""
|
|
return self.filter(orderproduct__isnull=True)
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self:
|
|
"""Filter products for a vendor that are not in any orders."""
|
|
return self.for_vendor(vendor).filter(orderproduct__isnull=True)
|
|
|
|
def mark_inactive(self) -> int:
|
|
"""Bulk update products to inactive status. Returns count updated."""
|
|
return self.update(is_active=False)
|
|
|
|
def mark_for_deletion(self, marker: str = "SCHON_DELETED_PRODUCT") -> int:
|
|
"""Mark products for deletion by setting description. Returns count updated."""
|
|
return self.update(description=marker)
|
|
|
|
def marked_for_deletion(self, marker: str = "SCHON_DELETED_PRODUCT") -> Self:
|
|
"""Get products marked for deletion."""
|
|
return self.filter(description__exact=marker)
|
|
|
|
def delete_with_related(self, batch_size: int = 5000) -> int:
|
|
"""
|
|
Delete products in batches along with their related objects.
|
|
|
|
Returns the total number of products deleted.
|
|
"""
|
|
from engine.core.models import AttributeValue, ProductImage
|
|
|
|
total_deleted = 0
|
|
while True:
|
|
batch_ids = list(self.values_list("pk", flat=True)[:batch_size])
|
|
if not batch_ids:
|
|
break
|
|
|
|
AttributeValue.objects.filter(product_id__in=batch_ids).delete()
|
|
ProductImage.objects.filter(product_id__in=batch_ids).delete()
|
|
deleted_count, _ = self.model.objects.filter(pk__in=batch_ids).delete()
|
|
total_deleted += deleted_count
|
|
|
|
return total_deleted
|
|
|
|
|
|
class ProductManager(MultilingualManager["Product"]):
|
|
"""Manager for Product model with custom queryset methods."""
|
|
|
|
_queryset_class = ProductQuerySet
|
|
|
|
def get_queryset(self) -> ProductQuerySet:
|
|
return ProductQuerySet(self.model, using=self._db)
|
|
|
|
def available(self) -> ProductQuerySet:
|
|
return self.get_queryset().available()
|
|
|
|
def available_in_stock(self) -> ProductQuerySet:
|
|
return self.get_queryset().available_in_stock()
|
|
|
|
def with_related(self) -> ProductQuerySet:
|
|
return self.get_queryset().with_related()
|
|
|
|
def stale(self) -> ProductQuerySet:
|
|
return self.get_queryset().stale()
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> ProductQuerySet:
|
|
return self.get_queryset().for_vendor(vendor)
|
|
|
|
def not_in_orders(self) -> ProductQuerySet:
|
|
return self.get_queryset().not_in_orders()
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> ProductQuerySet:
|
|
return self.get_queryset().for_vendor_not_in_orders(vendor)
|
|
|
|
|
|
class StockQuerySet(QuerySet["Stock"]):
|
|
"""Custom QuerySet for Stock with vendor-related operations."""
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> Self:
|
|
"""Filter stocks for a specific vendor."""
|
|
return self.filter(vendor=vendor)
|
|
|
|
def not_in_orders(self) -> Self:
|
|
"""Filter stocks whose products are not in any orders."""
|
|
return self.filter(product__orderproduct__isnull=True)
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self:
|
|
"""Filter stocks for a vendor whose products are not in orders."""
|
|
return self.for_vendor(vendor).filter(product__orderproduct__isnull=True)
|
|
|
|
|
|
class StockManager(models.Manager["Stock"]):
|
|
"""Manager for Stock model with custom queryset methods."""
|
|
|
|
def get_queryset(self) -> StockQuerySet:
|
|
return StockQuerySet(self.model, using=self._db)
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> StockQuerySet:
|
|
return self.get_queryset().for_vendor(vendor)
|
|
|
|
def not_in_orders(self) -> StockQuerySet:
|
|
return self.get_queryset().not_in_orders()
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> StockQuerySet:
|
|
return self.get_queryset().for_vendor_not_in_orders(vendor)
|
|
|
|
|
|
class AttributeValueQuerySet(QuerySet["AttributeValue"]):
|
|
"""Custom QuerySet for AttributeValue with vendor-related operations."""
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> Self:
|
|
"""Filter attribute values for products from a specific vendor."""
|
|
return self.filter(product__stocks__vendor=vendor)
|
|
|
|
def not_in_orders(self) -> Self:
|
|
"""Filter attribute values whose products are not in any orders."""
|
|
return self.filter(product__orderproduct__isnull=True)
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self:
|
|
"""Filter attribute values for a vendor whose products are not in orders."""
|
|
return self.for_vendor(vendor).filter(product__orderproduct__isnull=True)
|
|
|
|
|
|
class AttributeValueManager(models.Manager["AttributeValue"]):
|
|
"""Manager for AttributeValue model with custom queryset methods."""
|
|
|
|
def get_queryset(self) -> AttributeValueQuerySet:
|
|
return AttributeValueQuerySet(self.model, using=self._db)
|
|
|
|
def for_vendor(self, vendor: "Vendor") -> AttributeValueQuerySet:
|
|
return self.get_queryset().for_vendor(vendor)
|
|
|
|
def not_in_orders(self) -> AttributeValueQuerySet:
|
|
return self.get_queryset().not_in_orders()
|
|
|
|
def for_vendor_not_in_orders(self, vendor: "Vendor") -> AttributeValueQuerySet:
|
|
return self.get_queryset().for_vendor_not_in_orders(vendor)
|