from collections import defaultdict from datetime import timezone from decimal import Decimal, InvalidOperation from django.apps import apps from django.shortcuts import get_object_or_404, render,redirect from django.views import View from codesofy.custom_config import get_local_date, set_user_profile,convert_to_decimal,get_privilleges,get_standard_text_input,get_unique_text,is_authorized,set_menu_items,get_global_master_details from codesofy.master_details import BillingType, Brand, DateFilter, DensityUnit, DrugType, NumberFormatException,NegativeInputNumberException, UOM, PerPageSelector, Route,ProductType, TemperatureUnit from constants.general_const import ActiveStatus from core.models import Language from core.utils import _get_language_from_request, get_api_message from cropmanagement.models import Crop from customermanagement.models import Customer from products.serializers import ProductCatalogSerializer, ProductCropMappingLiteSerializer from usermanagement.models import UserProfile from utils.product_utils import dict_diff, product_to_dict from .models import CatelogTranslation, Product, ProductCropMapping from django.forms.models import model_to_dict from django.contrib import messages from django.db import transaction, IntegrityError from django.db.models import RestrictedError from django.views.decorators.cache import cache_control from django.utils.decorators import method_decorator import math from django.core.paginator import Paginator from django.http import HttpResponseBadRequest, JsonResponse import json from django.db.models import Max from PIL import Image from io import BytesIO from django.utils.timezone import now from django.contrib import messages from django.db import transaction, IntegrityError, models from django.db.models.deletion import RestrictedError from django.core.exceptions import ObjectDoesNotExist from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status, permissions from rest_framework.permissions import AllowAny from django.db.models.fields.files import FieldFile from .models import ProductHistory from rest_framework.permissions import IsAuthenticated MAX_UPLOAD_BYTES = 512 * 1024 # 512 KB menu_item = "product_registry" class HasCustomerProfile(permissions.BasePermission): """ Allow access only if: - request.user is a Customer instance (we set this in CustomerJWTAuthentication), OR - request.user has a 'customer' attribute (User -> Customer OneToOne) """ def has_permission(self, request, view): user = getattr(request, "user", None) if user is None: return False # If the authentication returned a Customer model instance directly if isinstance(user, Customer): return True # Otherwise, check for OneToOne relation on Django User (user.customer) return getattr(user, "customer", None) is not None def set_sub_menu_item(sub_menu_item,context): context = set_menu_items(menu_item,sub_menu_item,context) return context def get_local_master_details(): context = get_global_master_details() return context def get_master_details_for_add_update_product(): context = get_local_master_details() brand_list = Brand.to_list() default_temperature_list = TemperatureUnit.to_list() default_density_list = DensityUnit.to_list() context["default_temperature_list"] = default_temperature_list context["default_density_list"] = default_density_list context["product_type_list"] = ProductType.to_list() context["brand_list"] = brand_list context["status_list"] = ActiveStatus.to_list() return context def get_master_details_for_details_category(): context = get_local_master_details() context["product_type_list"] = ProductType.to_list_for_reports() context["brand_list"] = Brand.to_list_for_reports() return context def create_new_custom_product_code(): try: product_codes = Product.objects.values_list('code', flat=True) max_custom_product_code = max(int(code) for code in product_codes if code.isdigit()) new_custom_product_code = max_custom_product_code + 1 if new_custom_product_code < 10: new_custom_product_code = '000' + str(new_custom_product_code) elif new_custom_product_code < 100: new_custom_product_code = '00' + str(new_custom_product_code) elif new_custom_product_code < 1000: new_custom_product_code = '0' + str(new_custom_product_code) else: new_custom_product_code = str(new_custom_product_code) return new_custom_product_code except ValueError: return "0001" def validate_inputs(request): item_type_string = request.POST.get('item_type') item_type = ProductType(item_type_string).value unique_item_name = request.POST.get('unique_item_name') item_category_id = request.POST.get('item_category_id') item_brand_id = request.POST.get('item_brand_id') custom_item_code = request.POST.get('custom_item_code') store_id = request.POST.get('store_id') uom_string= request.POST.get('uom') uom = UOM(uom_string).value if (not item_type) or (not unique_item_name) or (not item_category_id) or (not item_brand_id) or (not store_id) or (not uom): #or (not custom_item_code) messages.error(request,"Backend Validation Failed! Please fill the required fileds") return False elif unique_item_name.strip()=='' :#or custom_item_code.strip()=='' messages.error(request,"Backend Validation Failed! Please fill the required fileds") return False # elif ' ' in custom_item_code.strip(): # messages.error(request,"Backend Validation Failed! Custom Item Code cannot contain spaces") # return False # elif len(custom_item_code) != 4: # messages.error(request,"Backend Validation Failed! Custom Item Code should contain 4 characters") # return False else: return True def validate_inputs_for_update(request): product_type_string = request.POST.get('product_type') product_type = ProductType(product_type_string).value unique_item_name = request.POST.get('unique_item_name') item_category_id = request.POST.get('item_category_id') item_brand_id = request.POST.get('item_brand_id') custom_item_code = request.POST.get('custom_item_code') store_id = request.POST.get('store_id') uom_string= request.POST.get('uom') uom = UOM(uom_string).value if (not product_type) or (not unique_item_name) or (not item_category_id) or (not item_brand_id) or (not store_id) or (not uom) or (not custom_item_code): messages.error(request,"Backend Validation Failed! Please fill the required fileds") return False elif unique_item_name.strip()=='' or custom_item_code.strip()=='': messages.error(request,"Backend Validation Failed! Please fill the required fileds") return False elif ' ' in custom_item_code.strip(): messages.error(request,"Backend Validation Failed! Custom Item Code cannot contain spaces") return False elif len(custom_item_code) != 4: messages.error(request,"Backend Validation Failed! Custom Item Code should contain 4 characters") return False else: return True def build_product_payload(instance, request): # 1) Start with ONLY concrete fields -> FKs come out as *_id, not objects concrete_field_names = [f.name for f in instance._meta.fields] data = model_to_dict(instance, fields=concrete_field_names) # 2) Convert any File/Image fields to absolute URLs for f in instance._meta.fields: if isinstance(f, (models.FileField, models.ImageField)): fileval = getattr(instance, f.name) data[f.name] = request.build_absolute_uri(fileval.url) if fileval else None # 3) OPTIONAL: add human-readable details for FK fields (won’t break JSON) for f in instance._meta.fields: # Many-to-one or one-to-one are FK-like if isinstance(f, (models.ForeignKey, models.OneToOneField)): rel_obj = getattr(instance, f.name, None) rel_id = getattr(instance, f"{f.name}_id", None) if rel_id is not None: # minimal, JSON-safe detail data[f"{f.name}_detail"] = { "id": rel_id, "repr": str(rel_obj) if rel_obj else None, } # 4) OPTIONAL: include display labels for choice fields (if any) for f in instance._meta.fields: if f.choices: # Django auto-creates get__display getter = getattr(instance, f"get_{f.name}_display", None) if callable(getter): data[f"{f.name}_display"] = getter() return data def get_master_details_for_add_update_catelog_translation(): context = get_local_master_details() context["language_list"] = Language.objects.order_by("name") context["product_list"] = Product.objects.order_by("name") context["status_list"] = ActiveStatus.to_list() return context def get_master_details_for_catelog_translation_list(): context = get_local_master_details() context["language_list"] = Language.objects.order_by("name") context["product_list"] = Product.objects.order_by("name") return context def get_master_details_for_catelog_translation_matrix(): context = get_local_master_details() context["language_list"] = Language.objects.order_by("name") context["product_list"] = Product.objects.order_by("name") return context def get_local_master_details_for_add_update_cropmapping(): context = get_local_master_details() product_list = Product.objects.all() crop_list = Crop.objects.all() mapping_status = ActiveStatus.to_list() context["mapping_status_list"] = mapping_status context['product_list'] = product_list context['crop_list'] = crop_list return context def get_local_master_details_for_cropmapping_details(): context = get_local_master_details() product_list = Product.objects.all() mapping_status = ActiveStatus.to_list_for_reports() context["mapping_status_list"] = mapping_status context['product_list'] = product_list return context def find_reverse_dependencies(instance, sample_per_relation: int = 3): """ Return a dict of {relation_label: [pk,...]} for reverse related objects that currently exist referencing `instance`. Uses model metadata, so you don't have to hardcode table names. """ blockers = {} # Walk all reverse relations declared on the model for rel in instance._meta.related_objects: # rel is a ManyToOneRel / ManyToManyRel / OneToOneRel (reverse side) accessor = rel.get_accessor_name() # e.g. "order_items", "results" manager = getattr(instance, accessor, None) if manager is None: continue # For O2O, manager is a single object accessor; handle safely try: if rel.one_to_one: try: obj = manager # If it exists, accessing .pk will not raise _ = obj.pk blockers[rel.related_model._meta.label] = [obj.pk] except rel.related_model.DoesNotExist: pass continue except AttributeError: # Older Django or unusual relation – skip continue # For reverse FK/M2M, manager is a RelatedManager try: qs = manager.all() except Exception: continue # Only flag if there are rows if qs.exists(): # Sample a few PKs for the message blockers[rel.related_model._meta.label] = list(qs.values_list("pk", flat=True)[:sample_per_relation]) return blockers @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class AddProductView(View): this_feature = "add_product" sub_menu_item = "add_product" def get(self, request): context = get_master_details_for_add_update_product() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) return render(request, 'add-product.html', context) def post(self, request): context = get_master_details_for_add_update_product() context["old_input_field_values"] = request.POST user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) uploaded = request.FILES.get('image') # None if not provided # ——— Validation (optional but recommended) ——— if uploaded: # size limit if uploaded.size > MAX_UPLOAD_BYTES: return HttpResponseBadRequest("Image too large. Max 512 KB.") # content type gate if uploaded.content_type not in ("image/png",): return HttpResponseBadRequest("PNG only.") # extra safety: verify it is really a PNG try: img_bytes = uploaded.read() uploaded.seek(0) # reset pointer for Django to save later with Image.open(BytesIO(img_bytes)) as im: if im.format != 'PNG': return HttpResponseBadRequest("PNG only.") except Exception: return HttpResponseBadRequest("Invalid image file.") product_type_string = request.POST.get('product_type') product_type = ProductType(product_type_string).value product_name = get_standard_text_input(request.POST.get('product_name')) product_brand = request.POST.get('product_brand') custom_product_code = request.POST.get('custom_product_code') description = request.POST.get('description') is_manual_integration = request.POST.get('is_manual_integration') is_QR_integration = request.POST.get('is_QR_integration') is_bluetooth_integration = request.POST.get('is_bluetooth_integration') is_manual_result_reading = request.POST.get('is_manual_result_reading') is_qr_result_reading = request.POST.get('is_qr_result_reading') is_bluetooth_result_reading = request.POST.get('is_bluetooth_result_reading') model_id = request.POST.get('model_id') catelog_link = request.POST.get('catelog_link').strip() default_temperature = request.POST.get('default_temperature') default_density = request.POST.get('default_density') available_for_device_pairing = str(request.POST.get('available_for_device_pairing')).lower() in ("1", "true", "on", "yes") try: with transaction.atomic(): if Product.objects.filter(name=product_name.upper()).exists(): messages.error(request, "Unique Product Name already exists") return render(request, 'add-product.html', context) if Product.objects.filter(code=custom_product_code).exists(): messages.error(request, "Custom PRoduct Code already exists") return render(request, 'add-product.html', context) product = Product.objects.create( name = product_name, code = custom_product_code, type = product_type, brand = product_brand, image = uploaded, description = description, is_manual_integration = is_manual_integration, is_QR_integration = is_QR_integration, is_bluetooth_integration = is_bluetooth_integration, is_manual_result_reading = is_manual_result_reading, is_qr_result_reading = is_qr_result_reading, is_bluetooth_result_reading = is_bluetooth_result_reading, created_user = user_profile, updated_user = user_profile, model_id = model_id, catelog_link = catelog_link, default_temperature = default_temperature, default_density = default_density, available_for_device_pairing = available_for_device_pairing, ) messages.success(request, "New product is created successfully") return redirect('view-product', id=product.pk) except Exception as exp: messages.error(request, f"Unexpected Error: {exp}") messages.error(request, "Error occurred in atomic operation. Contact the administrator.") return render(request, 'add-product.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class ViewProductView(View): this_feature = "view_product" sub_menu_item = "product_list" def get(self, request,id): context = get_local_master_details() user_profile = set_user_profile(request,context) if user_profile==None: return redirect('login') get_privilleges(user_profile,context) if not is_authorized(user_profile,self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item,context) try: with transaction.atomic(): if Product.objects.filter(pk=id).exists(): product = Product.objects.get(pk=id) context["product"] = product return render(request, 'view-product.html', context) else: return redirect ('product-list') except Exception or IntegrityError as exp: print (exp) messages.error( request, "Error Occured in atomic operation. Contact the System Administrator") return redirect ('product-list') @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class ProductListView(View): this_feature = "product_list" sub_menu_item = "product_list" def get(self, request): context = get_local_master_details() user_profile = set_user_profile(request,context) if user_profile==None: return redirect('login') get_privilleges(user_profile,context) if not is_authorized(user_profile,self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item,context) product_name = request.GET.get('product_name') model_name_id = request.GET.get('model_name_id') device_model = request.GET.get('device_model') per_page_count = request.GET.get('per_page_count') page_number = request.GET.get('page_number') start_index = request.GET.get('start_index') if not page_number: page_number = 1 if not per_page_count: per_page_count = 10 if start_index: page_number = math.ceil(int(start_index)/int(per_page_count)) context["default_per_page_count"] = per_page_count filter_kwargs = {} if device_model: device_model = get_unique_text(device_model) context["default_device_model"] = device_model filter_kwargs['model_id__icontains']= device_model if model_name_id: model_name_id = get_unique_text(model_name_id) context["default_model_name_id"] = model_name_id filter_kwargs['code__icontains']= model_name_id if product_name: product_name = get_unique_text(product_name) context["default_product_name"] = product_name filter_kwargs['name__icontains']= product_name product_list = Product.objects.filter(**filter_kwargs).order_by('name') if len(product_list)>0: context["has_entries"] = True else: context["has_entries"] = False paginator = Paginator(product_list,per_page_count) page = Paginator.get_page(paginator,page_number) page_list = list(paginator.get_elided_page_range(page_number, on_each_side=1)) context["page"] = page context["page_list"] = page_list return render(request, 'product-list.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class UpdateProductView(View): this_feature = "update_product" sub_menu_item = "product_list" def get(self, request, id): context = get_master_details_for_add_update_product() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') # Redirect to login if the user is not logged in. # Check user privileges and authorization get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') # Redirect if the user is not authorized. # Set submenu item context set_sub_menu_item(self.sub_menu_item, context) # Initialize empty input values for GET requests context["old_input_field_values"] = {} # Check if the item exists in the database if Product.objects.filter(pk=id).exists(): product = Product.objects.get(pk=id) context["product"] = product return render(request, 'update-product.html', context) else: messages.error(request, "Product doesn't exist") # Error if the item does not exist. return redirect('product-list') # Redirect to the item list page if the item doesn't exist. def post(self, request, id): context = get_master_details_for_add_update_product() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) context["old_input_field_values"] = request.POST # Helpers def _s(val): return (val or "").strip() def _as_bool(name): # Accept common truthy values for checkboxes/switches return str(request.POST.get(name)).lower() in ("1", "true", "on", "yes") # Parse inputs uploaded = request.FILES.get('image') product_type_str = request.POST.get('product_type') try: # If you store enum "value" in DB, keep .value; otherwise use .name product_type = ProductType(product_type_str).value except Exception: messages.error(request, "Invalid product type.") return render(request, 'update-product.html', context) product_name = get_standard_text_input(request.POST.get('product_name')) product_brand = request.POST.get('product_brand') description = request.POST.get('description') is_manual_integration = _as_bool('is_manual_integration') is_QR_integration = _as_bool('is_QR_integration') is_bluetooth_integration = _as_bool('is_bluetooth_integration') is_manual_result_reading = _as_bool('is_manual_result_reading') is_qr_result_reading = _as_bool('is_qr_result_reading') is_bluetooth_result_reading = _as_bool('is_bluetooth_result_reading') custom_product_code = request.POST.get('custom_product_code') # unused? (keep if needed later) status = request.POST.get('status') model_id = request.POST.get('model_id') available_for_device_pairing = _as_bool('available_for_device_pairing') catelog_link = _s(request.POST.get('catelog_link')) default_temperature = _s(request.POST.get('default_temperature')) default_density = _s(request.POST.get('default_density')) try: with transaction.atomic(): # Lock row during update original_product = ( Product.objects.select_for_update().get(pk=id) ) context["original_product"] = original_product # Snapshot BEFORE (for your audit utilities) before = product_to_dict(original_product) changed = False # Assign fields only if changed updates = { "code": custom_product_code, "catelog_link": catelog_link, "model_id": model_id, "type": product_type, "name": product_name, "brand": product_brand, "description": description, "is_manual_integration": is_manual_integration, "is_QR_integration": is_QR_integration, "is_bluetooth_integration": is_bluetooth_integration, "is_manual_result_reading": is_manual_result_reading, "is_qr_result_reading": is_qr_result_reading, "is_bluetooth_result_reading": is_bluetooth_result_reading, "available_for_device_pairing":available_for_device_pairing, "status": status, "default_temperature": default_temperature, "default_density": default_density, "updated_user": user_profile, "updated_At": now(), } for field, value in updates.items(): if getattr(original_product, field) != value: setattr(original_product, field, value) changed = True # Handle image replacement if uploaded: # Delete old file if exists, then set new file if original_product.image: original_product.image.delete(save=False) original_product.image = uploaded changed = True if not changed: messages.info(request, "No changes detected.") return render(request, 'update-product.html', context) # Pass metadata for signals/auditing if you use them original_product._changed_by = user_profile original_product._change_notes = "Updated via UI" original_product.save() # AFTER snapshot + (optional) diff usage after = product_to_dict(original_product) # If you want, you can still check/record diff here: # diff = dict_diff(before, after) messages.success(request, "Product details updated successfully.") return redirect('view-product', id=original_product.pk) except Product.DoesNotExist: messages.error(request, "Product not found.") return redirect('product-details') # or your list route except Exception as exp: messages.error(request, f"{exp}") messages.error(request, "Error occurred in atomic operation. Contact the System Administrator") return render(request, 'update-product.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class DeleteProductView(View): this_feature = "delete_item" def post(self, request): context = {} user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') code = request.POST.get('custom_product_code') if not code: messages.error(request, "Missing product code.") return redirect('product-list') product = get_object_or_404(Product, code=code) # ---- Pre-check: block deletion if referenced anywhere (except history) ---- in_use, blockers = self._is_in_use(product) if in_use: msg = "Deletion restricted. This product is referenced by: " + ", ".join(sorted(blockers)) messages.error(request, msg) return redirect('view-product', product.id) try: # write a final history snapshot BEFORE deleting self._write_final_delete_history(product, user_profile, request) with transaction.atomic(): product.delete() messages.success(request, "Product deleted.") return redirect('product-list') except RestrictedError: messages.error(request, "Deletion restricted. This product is in use.") return redirect('view-product', product.id) except IntegrityError: messages.error(request, "Deletion failed due to related data constraints.") return redirect('view-product', product.id) except Exception as e: messages.error(request, f"Unexpected error during delete: {e}") return redirect('view-product', product.id) # -------- helpers -------- def _is_in_use(self, obj): """ Inspect reverse relations and return (True, {model_names}) if any related rows exist, IGNORING ProductHistory relations. """ # Resolve ProductHistory safely (replace 'your_app_label' with the real app label) ProductHistory = apps.get_model('products', 'ProductHistory') IGNORE_MODELS = {ProductHistory} blockers = set() for rel in obj._meta.related_objects: if rel.related_model in IGNORE_MODELS: continue accessor = rel.get_accessor_name() try: manager_or_obj = getattr(obj, accessor) except AttributeError: continue if rel.one_to_one: try: getattr(obj, accessor) # raises DoesNotExist if absent blockers.add(rel.related_model._meta.verbose_name_plural.title()) except rel.related_model.DoesNotExist: pass continue try: if manager_or_obj.all().exists(): blockers.add(rel.related_model._meta.verbose_name_plural.title()) except Exception: blockers.add(rel.related_model._meta.verbose_name_plural.title()) return (len(blockers) > 0, blockers) def _write_final_delete_history(self, product, user_profile, request): ProductHistory = apps.get_model('products', 'ProductHistory') # <-- use your app label has_product_code = any(f.name == "product_code" for f in ProductHistory._meta.get_fields()) version_qs = ( ProductHistory.objects.filter(product_code=product.code) if has_product_code else ProductHistory.objects.filter(product=product) ) last_ver = version_qs.aggregate(v=Max("version")).get("v") or 0 payload = self._serialize_product(product, request) create_kwargs = dict( product=product, version=last_ver + 1, snapshot=payload, diff={"action": "delete"}, changed_by=user_profile, change_notes="Product deleted", ) if has_product_code: create_kwargs["product_code"] = product.code ProductHistory.objects.create(**create_kwargs) def _serialize_product(self, product, request): """ JSON-serializable product snapshot (handles ImageField & FKs). """ def img_url(p): try: return request.build_absolute_uri(p.image.url) if p.image else None except Exception: return None return { "id": product.id, "code": product.code, "name": product.name, "brand": product.brand, "type": product.type, "is_manual_integration": product.is_manual_integration, "is_QR_integration": product.is_QR_integration, "is_bluetooth_integration": product.is_bluetooth_integration, "is_manual_result_reading": product.is_manual_result_reading, "is_qr_result_reading": product.is_qr_result_reading, "is_bluetooth_result_reading": product.is_bluetooth_result_reading, "image": img_url(product), "description": product.description, "status": product.status, "created_At": product.created_At.isoformat() if product.created_At else None, "updated_At": product.updated_At.isoformat() if product.updated_At else None, "created_user_id": product.created_user_id, "updated_user_id": product.updated_user_id, "model_id": product.model_id, "catelog_link": product.catelog_link, "default_temperature": product.default_temperature, "default_density": product.default_density, } @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class GetAllDeviceView(APIView): permission_classes = [AllowAny] def get(self, request): try: product_type = request.query_params.get("type") qs = ( Product.objects .filter( status=ActiveStatus.ACTIVE.value, available_for_device_pairing=True ).order_by("code") ) if product_type: qs = qs.filter(type=product_type) language = _get_language_from_request(request) translation_map = {} if language: translations = CatelogTranslation.objects.filter( language=language, product_id__in=qs.values_list("id", flat=True) ).values_list("product_id", "catelog_link") translation_map = { product_id: link for product_id, link in translations } serializer = ProductCatalogSerializer( qs, many=True, context={ "request": request, "translation_map": translation_map, }, ) return JsonResponse( { "success": True, "message": get_api_message("DEVICE_LIST", request), "code": "DEVICE_LIST", "data": serializer.data, }, status=200, ) except Exception as e: return JsonResponse( { "success": False, "message": get_api_message("DEVICE_LIST_FAILED", request), "code": "DEVICE_LIST_FAILED", "errors": {"exception": [str(e)]}, }, status=500, ) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class GetDeviceDetailView(APIView): permission_classes = [AllowAny] def get(self, request, id): try: device = ( Product.objects .filter(id=id, status=ActiveStatus.ACTIVE.value) .values() .first() ) if not device: return JsonResponse( { "success": False, "message": get_api_message("DEVICE_NOT_FOUND", request), "code": "DEVICE_NOT_FOUND", "errors": {"id": [f"No active device with id {id}."]}, }, status=404, ) return JsonResponse( { "success": True, "message": get_api_message("DEVICE_FETCHED", request), "code": "DEVICE_FETCHED", "data": device, # object directly }, status=200, ) except Exception as e: return JsonResponse( { "success": False, "message": get_api_message("DEVICE_FETCH_FAILED", request), "code": "DEVICE_FETCH_FAILED", "errors": {"exception": [str(e)]}, }, status=500, ) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class GetDeviceDetailByCodeView(APIView): permission_classes = [AllowAny] def get(self, request): code = request.query_params.get("model_name_id") or request.query_params.get("code") model_id = request.query_params.get("device_model") or request.query_params.get("model_id") if not code or not model_id: return JsonResponse( { "success": False, "message": get_api_message("MISSING_QUERY_PARAMS", request), "code": "MISSING_QUERY_PARAMS", "errors": { "model_name_id": ["Required"], "device_model": ["Required"], }, }, status=400, ) try: product = Product.objects.filter( code=code, model_id=model_id, status=ActiveStatus.ACTIVE.value, ).first() if not product: return JsonResponse( { "success": False, "message": get_api_message("DEVICE_NOT_FOUND", request), "code": "DEVICE_NOT_FOUND", "errors": {"pair": [f"(code={code}, model_id={model_id}) not found."]}, }, status=404, ) payload = build_product_payload(product, request) # your existing helper return JsonResponse( { "success": True, "message": get_api_message("DEVICE_FETCHED", request), "code": "DEVICE_FETCHED", "data": payload, }, status=200, ) except Exception as e: return JsonResponse( { "success": False, "message": get_api_message("DEVICE_FETCH_FAILED", request), "code": "DEVICE_FETCH_FAILED", "errors": {"exception": [str(e)]}, }, status=500, ) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class GetProductCatelogView(APIView): permission_classes = [IsAuthenticated, HasCustomerProfile] def get(self, request): try: product_type = request.query_params.get("type") brand = request.query_params.get("brand") qs = ( Product.objects .filter(status=ActiveStatus.ACTIVE.value) .only("id", "code", "name", "brand", "type", "description", "model_id", "catelog_link") .order_by("code") ) if product_type: qs = qs.filter(type=product_type) if brand: qs = qs.filter(brand=brand) language = _get_language_from_request(request) translation_map = {} if language: translations = CatelogTranslation.objects.filter( language=language, product_id__in=qs.values_list("id", flat=True) ).values_list("product_id", "catelog_link") translation_map = {product_id: link for product_id, link in translations} serializer = ProductCatalogSerializer( qs, many=True, context={ "request": request, "translation_map": translation_map, }, ) return JsonResponse( { "success": True, "message": get_api_message("PRODUCT_LIST", request), "code": "PRODUCT_LIST", "data": serializer.data, }, status=200, ) except Exception as e: return JsonResponse( { "success": False, "message": get_api_message("PRODUCT_FETCH_FAILED", request), "code": "PRODUCT_FETCH_FAILED", "errors": {"exception": [str(e)]}, }, status=500, ) # wile/products/views.py @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class AddCatelogTranslationView(View): this_feature = "add_catelog_translation" sub_menu_item = "add_catelog_translation" def get(self, request): context = get_master_details_for_add_update_catelog_translation() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) return render(request, "add-catelog-translation.html", context) def post(self, request): context = get_master_details_for_add_update_catelog_translation() context["old_input_field_values"] = request.POST user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) product_id = (request.POST.get("product_id") or "").strip() language_id = (request.POST.get("language_id") or "").strip() catelog_link = (request.POST.get("catelog_link") or "").strip() catelog_link = " ".join(catelog_link.split()) if not product_id or not language_id or not catelog_link: messages.error(request, "Product, language, and catelog link are required.") return render(request, "add-catelog-translation.html", context) product = Product.objects.filter(pk=product_id).first() if not product: messages.error(request, "Invalid product selected.") return render(request, "add-catelog-translation.html", context) language = Language.objects.filter(pk=language_id).first() if not language: messages.error(request, "Invalid language selected.") return render(request, "add-catelog-translation.html", context) # Duplicate checks (safe + user friendly) if CatelogTranslation.objects.filter(product=product, language=language).exists(): messages.error(request, f'A translation already exists for "{product.name}" in "{language.name}".') return render(request, "add-catelog-translation.html", context) try: with transaction.atomic(): obj = CatelogTranslation( product=product, language=language, catelog_link=catelog_link, created_by=user_profile, updated_by=user_profile, ) obj.save() messages.success(request, "Catelog translation created successfully.") return redirect("view-catelog-translation", id=obj.id) except IntegrityError as e: messages.error(request, f"DB error: {str(e)}") return render(request, "add-catelog-translation.html", context) except Exception: messages.error(request, "Unexpected error. Contact the system administrator.") return render(request, "add-catelog-translation.html", context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class ViewCatelogTranslationView(View): this_feature = "view_catelog_translation" sub_menu_item = "catelog_translation_list" def get(self, request, id): context = get_master_details_for_add_update_catelog_translation() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) obj = get_object_or_404( CatelogTranslation.objects.select_related("product", "language"), pk=id ) context["obj"] = obj return render(request, "view-catelog-translation.html", context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class UpdateCatelogTranslationView(View): this_feature = "update_catelog_translation" sub_menu_item = "catelog_translation_list" def get(self, request, id): context = get_master_details_for_add_update_catelog_translation() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) obj = get_object_or_404(CatelogTranslation, pk=id) context["obj"] = obj # For selected dropdown values (same style as your templates) context["old_input_field_values"] = { "product_id": str(obj.product_id), "language_id": str(obj.language_id), "catelog_link": obj.catelog_link, } return render(request, "update-catelog-translation.html", context) def post(self, request, id): context = get_master_details_for_add_update_catelog_translation() context["old_input_field_values"] = request.POST user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) obj = get_object_or_404(CatelogTranslation, pk=id) product_id = (request.POST.get("product_id") or "").strip() language_id = (request.POST.get("language_id") or "").strip() catelog_link = (request.POST.get("catelog_link") or "").strip() catelog_link = " ".join(catelog_link.split()) if not product_id or not language_id or not catelog_link: messages.error(request, "Product, language, and catelog link are required.") context["obj"] = obj return render(request, "update-catelog-translation.html", context) product = Product.objects.filter(pk=product_id).first() if not product: messages.error(request, "Invalid product selected.") context["obj"] = obj return render(request, "update-catelog-translation.html", context) language = Language.objects.filter(pk=language_id).first() if not language: messages.error(request, "Invalid language selected.") context["obj"] = obj return render(request, "update-catelog-translation.html", context) # Duplicate checks excluding current record if CatelogTranslation.objects.filter(product=product, language=language).exclude(pk=obj.id).exists(): messages.error(request, f'A translation already exists for "{product.name}" in "{language.name}".') context["obj"] = obj return render(request, "update-catelog-translation.html", context) try: with transaction.atomic(): obj.product = product obj.language = language obj.catelog_link = catelog_link obj.updated_by = user_profile obj.save() messages.success(request, "Catelog translation updated successfully.") return redirect("view-catelog-translation", id=obj.id) except IntegrityError as e: messages.error(request, f"DB error: {str(e)}") context["obj"] = obj return render(request, "update-catelog-translation.html", context) except Exception: messages.error(request, "Unexpected error. Contact the system administrator.") context["obj"] = obj return render(request, "update-catelog-translation.html", context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class DeleteCatelogTranslationView(View): this_feature = "delete_catelog_translation" sub_menu_item = "catelog_translation_list" def post(self, request): context = get_master_details_for_catelog_translation_list() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") obj_id = (request.POST.get("translation_id") or "").strip() if not obj_id: messages.error(request, "Invalid request.") return redirect("catelog-translation-list") obj = CatelogTranslation.objects.filter(pk=obj_id).first() if not obj: messages.error(request, "Record not found.") return redirect("catelog-translation-list") try: with transaction.atomic(): obj.delete() messages.success(request, "Catelog translation deleted successfully.") return redirect("catelog-translation-list") except RestrictedError: messages.error(request, "This record is in use and cannot be deleted.") return redirect("catelog-translation-list") except Exception: messages.error(request, "Unexpected error. Contact the system administrator.") return redirect("catelog-translation-list") @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class CatelogTranslationListView(View): this_feature = "catelog_translation_list" sub_menu_item = "catelog_translation_list" def get(self, request): context = get_master_details_for_catelog_translation_list() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) # Filters product_id = (request.GET.get("product_id") or "").strip() language_id = (request.GET.get("language_id") or "").strip() if product_id: context["default_product_id"] = product_id if language_id: context["default_language_id"] = language_id per_page_count = request.GET.get('per_page_count') or PerPageSelector.get_default().value page_number = request.GET.get('page_number') or 1 start_index = request.GET.get('start_index') if start_index: page_number = math.ceil(int(start_index) / int(per_page_count)) filter_kwargs = {} if product_id and product_id.lower() != "all": filter_kwargs["product_id"] = product_id if language_id and language_id.lower() != "all": filter_kwargs["language_id"] = language_id qs = ( CatelogTranslation.objects .select_related("product", "language") .filter(**filter_kwargs) .order_by("-id") ) context["has_entries"] = qs.exists() paginator = Paginator(qs, per_page_count) page = Paginator.get_page(paginator, page_number) page_list = list(paginator.get_elided_page_range(page_number, on_each_side=1)) context["page"] = page context["page_list"] = page_list return render(request, 'catelog-translation-list.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class CatelogTranslationMatrixView(View): this_feature = "catelog_translation_matrix" sub_menu_item = "catelog_translation_matrix" def get(self, request): context = get_master_details_for_catelog_translation_matrix() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) # Optional filters product_type = (request.GET.get("type") or "").strip() brand = (request.GET.get("brand") or "").strip() products = Product.objects.all().order_by("name") if product_type: products = products.filter(type=product_type) if brand: products = products.filter(brand=brand) languages = Language.objects.order_by("name") # Build nested dict: translation_map[product_id][language_id] = catelog_link translation_map = {} rows = CatelogTranslation.objects.values_list("product_id", "language_id", "catelog_link") for product_id, language_id, link in rows: if product_id not in translation_map: translation_map[product_id] = {} translation_map[product_id][language_id] = link context["products"] = products context["languages"] = languages context["translation_map"] = translation_map return render(request, "catelog-translation-matrix.html", context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class AddCropMappingView(View): this_feature = "add_cropmapping" sub_menu_item = "add_cropmapping" def get(self, request): context = get_local_master_details_for_add_update_cropmapping() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) return render(request, "add-cropmapping.html", context) def post(self, request): context = get_local_master_details_for_add_update_cropmapping() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) # ---- Raw inputs (IDs for FKs expected from the form) ---- product_id = (request.POST.get("product") or "").strip() crop_id = (request.POST.get("crop") or "").strip() # scale_order_number = (request.POST.get("scale_order_number") or "").strip() universal_moisture_threshold = (request.POST.get("universal_moisture_threshold") or "").strip() # Keep what the user typed to re-fill the form on error def keep_form_values(): context["form_values"] = { "product": product_id, "crop": crop_id, "universal_moisture_threshold": universal_moisture_threshold, } # ---- Required fields check ---- if not product_id or not crop_id: messages.error(request, "Product and crop are required.") keep_form_values() return render(request, "add-cropmapping.html", context) # ---- Resolve FKs ---- try: product = Product.objects.get(pk=product_id) crop = Crop.objects.get(pk=crop_id) except ObjectDoesNotExist: messages.error(request, "Invalid product or crop selection.") keep_form_values() return render(request, "add-cropmapping.html", context) # ---- Parse optional numeric field (no range validation) ---- umt = None if universal_moisture_threshold: try: # Use Decimal to match DecimalField and avoid float quirks umt = Decimal(universal_moisture_threshold) except (InvalidOperation, ValueError): messages.error(request, "Moisture threshold must be a numeric value.") keep_form_values() return render(request, "add-cropmapping.html", context) # ---- Create with race-safety & unique-pair handling ---- try: with transaction.atomic(): mapping = ProductCropMapping.objects.create( product=product, crop=crop, universal_moisture_threshold=umt, created_user=user_profile, updated_user=user_profile, ) except IntegrityError: # UniqueConstraint(fields=['product','crop']) hit messages.error(request, "That product is already mapped to this crop.") keep_form_values() return render(request, "add-cropmapping.html", context) messages.success( request, f'Mapping created: "{mapping.product.name} ↔ {mapping.crop.name}".' ) return redirect("view-cropmapping", id=mapping.id) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class ViewCropMappingView(View): this_feature = "view_cropmapping" sub_menu_item = "cropmapping_details" def get(self, request, id): context = get_local_master_details_for_cropmapping_details() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) # Read-only fetch; no transaction needed mapping = get_object_or_404( ProductCropMapping.objects.select_related('product', 'crop'), pk=id ) context["mapping"] = mapping return render(request, 'view-cropmapping.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class CropMappingDetailsView(View): this_feature = "cropmapping_details" sub_menu_item = "cropmapping_details" def get(self, request): context = get_local_master_details_for_cropmapping_details() user_profile = set_user_profile(request,context) if user_profile==None: return redirect('login') get_privilleges(user_profile,context) if not is_authorized(user_profile,self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item,context) crop = request.GET.get('crop') product = request.GET.get('product') # scale_order_number = request.GET.get('scale_order_number') # is_active = request.GET.get('is_active') per_page_count = request.GET.get('per_page_count') page_number = request.GET.get('page_number') start_index = request.GET.get('start_index') if not page_number: page_number = 1 if not per_page_count: per_page_count = PerPageSelector.get_default().value if start_index: page_number = math.ceil(int(start_index)/int(per_page_count)) #filter logic applies here filter_kwargs = {} # if scale_order_number: # context["default_scale_order_number"] = scale_order_number # filter_kwargs['scale_order_number__icontains']= scale_order_number if product: # filter by product code (case-insensitive exact) filter_kwargs["product__code__iexact"] = product context ["default_product"] = product if crop: context["default_crop"] = crop filter_kwargs['crop__name']= crop # if is_active: # if is_active.lower()!="all": # context["default_is_active"] = ActiveStatus(is_active).value # filter_kwargs['is_active']= ActiveStatus(is_active).value mapping_list = ProductCropMapping.objects.filter(**filter_kwargs).order_by('-id') if len(mapping_list)>0: context["has_entries"] = True else: context["has_entries"] = False paginator = Paginator(mapping_list,per_page_count) page = Paginator.get_page(paginator,page_number) page_list = list(paginator.get_elided_page_range(page_number, on_each_side=1)) context["page"] = page context["page_list"] = page_list return render(request, 'cropmapping-details.html', context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class UpdateCropMappingView(View): this_feature = "update_cropmapping" sub_menu_item = "cropmapping_details" def get(self, request, id): context = get_local_master_details_for_add_update_cropmapping() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) # Load mapping (NOT Crop) mapping = get_object_or_404(ProductCropMapping, pk=id) context["mapping"] = mapping # Pre-fill form fields (useful if your template reads old_input_field_values) context["old_input_field_values"] = { "product": str(mapping.product_id), "crop": str(mapping.crop_id), # "scale_order_number": mapping.scale_order_number or "", "universal_moisture_threshold": ( "" if mapping.universal_moisture_threshold is None else str(mapping.universal_moisture_threshold) ), } return render(request, 'update-cropmapping.html', context) def post(self, request, id): context = get_local_master_details_for_add_update_cropmapping() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) mapping = ProductCropMapping.objects.filter(pk=id).select_related("product", "crop").first() if not mapping: messages.error(request, "Product–Crop mapping not found.") return render(request, 'update-cropmapping.html', context) # Read & normalize inputs (expecting **IDs** from the form) product_id = (request.POST.get("product") or "").strip() crop_id = (request.POST.get("crop") or "").strip() # scale_order_number = (request.POST.get("scale_order_number") or "").strip() universal_moisture_threshold = (request.POST.get("universal_moisture_threshold") or "").strip() # Preserve user input on error context["old_input_field_values"] = { "product": product_id, "crop": crop_id, # "scale_order_number": scale_order_number, "universal_moisture_threshold": universal_moisture_threshold, } # Basic required fields for FKs if not product_id or not crop_id: messages.error(request, "Product and crop are required.") return render(request, 'update-cropmapping.html', context) # Resolve FKs by primary key (IDs). If your template posts codes/names instead, # switch to filtering by unique fields (code__iexact/name__iexact) as shown earlier. try: product = Product.objects.get(pk=product_id) crop = Crop.objects.get(pk=crop_id) except (Product.DoesNotExist, Crop.DoesNotExist): messages.error(request, "Invalid product or crop selection.") return render(request, 'update-cropmapping.html', context) # Parse optional numeric field (Decimal to match DecimalField; no range enforcement here) umt = None if universal_moisture_threshold: try: umt = Decimal(universal_moisture_threshold) except (InvalidOperation, ValueError): messages.error(request, "Moisture threshold must be a numeric value.") return render(request, 'update-cropmapping.html', context) # Update safely; handle unique (product, crop) collisions try: with transaction.atomic(): mapping.product = product mapping.crop = crop # mapping.scale_order_number = scale_order_number mapping.universal_moisture_threshold = umt # can be None mapping.updated_user = user_profile mapping.updated_at = timezone.now() mapping.save() # may raise IntegrityError on unique constraint except IntegrityError: # Likely violates UniqueConstraint('product', 'crop') messages.error(request, "Another mapping already exists for that Product ↔ Crop pair.") return render(request, 'update-cropmapping.html', context) except Exception: messages.error(request, "Unexpected error. Please contact the system administrator.") return render(request, 'update-cropmapping.html', context) messages.success( request, f'Updated mapping: "{mapping.product.name} ↔ {mapping.crop.name}".' ) # Choose the redirect that matches your URLs: # return redirect('cropmapping-details') # if it’s an index/list page return redirect('view-cropmapping', id=mapping.id) # if you have a detail view @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class DeleteCropMappingView(View): this_feature = "delete_cropmapping" sub_menu_item = "cropmapping_details" def post(self, request): context = get_local_master_details_for_cropmapping_details() user_profile = set_user_profile(request, context) if user_profile is None: return redirect('login') get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect('unauthorized-access') set_sub_menu_item(self.sub_menu_item, context) mapping_id = (request.POST.get('mapping_id') or '').strip() if not mapping_id: messages.error(request, "No mapping id provided.") return redirect('cropmapping-details') # Fetch the mapping first (no lock yet) mapping = ProductCropMapping.objects.select_related("product", "crop").filter(pk=mapping_id).first() if not mapping: messages.error(request, "Product–Crop mapping not found.") return redirect('cropmapping-details') # 1) Pre-check for dependencies (reverse relations) blockers = find_reverse_dependencies(mapping) if blockers: # Build a friendly message listing what blocks the deletion lines = [] for label, ids in blockers.items(): lines.append(f"{label} (example ids: {', '.join(map(str, ids))})") msg = "Cannot delete: this mapping is referenced by other records → " + "; ".join(lines) messages.error(request, msg) return redirect('cropmapping-details') # 2) No blockers found; lock + delete atomically to prevent race try: with transaction.atomic(): # Lock the same row before deleting, to avoid a new reference being created mid-flight locked = ProductCropMapping.objects.select_for_update().get(pk=mapping.pk) # Double-check dependencies right before delete (TOCTOU safety) blockers_now = find_reverse_dependencies(locked) if blockers_now: lines = [] for label, ids in blockers_now.items(): lines.append(f"{label} (example ids: {', '.join(map(str, ids))})") msg = "Cannot delete (just referenced): " + "; ".join(lines) messages.error(request, msg) return redirect('cropmapping-details') product_name = locked.product.name crop_name = locked.crop.name locked.delete() messages.success(request, f'Mapping "{product_name} ↔ {crop_name}" deleted.') return redirect('cropmapping-details') except ProductCropMapping.DoesNotExist: messages.error(request, "Product–Crop mapping not found.") return redirect('cropmapping-details') except Exception as e: # Fallback: if your DB-level FKs are RESTRICT/PROTECT, this will still be safe. messages.error(request, f"Could not delete. {e}") return redirect('cropmapping-details') @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name="dispatch") class ProductCropMappingsByQuery(APIView): permission_classes = [AllowAny] def get(self, request): product_id = request.query_params.get("product_id") if not product_id: return JsonResponse( { "success": False, "message": get_api_message("MISSING_PRODUCT_ID", request), "code": "MISSING_PRODUCT_ID", "errors": {"product_id": ["This query parameter is required."]}, }, status=400, ) if not Product.objects.filter(pk=product_id).exists(): return JsonResponse( { "success": False, "message": get_api_message("PRODUCT_NOT_FOUND", request), "code": "PRODUCT_NOT_FOUND", "errors": {"product_id": [f"No product with id {product_id}."]}, }, status=404, ) qs = ( ProductCropMapping.objects .filter(product_id=product_id) .select_related("crop") .order_by("scale_order_number") ) data = ProductCropMappingLiteSerializer( qs, many=True, context={"request": request} ).data return JsonResponse( { "success": True, "message": get_api_message("MAPPINGS_FETCHED", request), "code": "MAPPINGS_FETCHED", "data": data, }, status=200, )