"""Authorized inventory foundation commands and the common posting primitive.""" import uuid from django.core.exceptions import ValidationError from django.db import transaction from apps.access.authorization import require_permission from apps.organization.models import Company, ServiceCenter from apps.parts.locking import persisted_pk from apps.parts.models import SparePart from .locking import inventory_context, lock_positions from .models import InventoryLocation, SerializedStockUnit, StockMovement, StockLedgerEntry, StockMovementUnit from .queries import _on_hand, USABLE_TYPES def text_value(value, *, required=False, maximum=2000): if not isinstance(value, str) or len(value.strip()) > maximum or (required and not value.strip()): raise ValidationError("Supply valid bounded text.") return value.strip() def quantity_value(value): if type(value) is not int or not 1 <= value <= 1000000000: raise ValidationError("Quantity must be a positive whole number, at most 1,000,000,000.") return value def check_revision(record, expected_revision): if expected_revision is not None and record.updated_at.isoformat() != expected_revision: raise ValidationError("The record changed; reload and review before retrying.") def create_location(*, actor, company, code, name, location_type, service_center=None): if location_type in ("TRANSIT", "CUSTODY"): raise ValidationError("Transit locations are managed by transfer workflows.") # A center-scoped actor need not have company-wide scope. Use a transient # authorization target while the company lock stabilizes its hierarchy. persisted_pk(company, Company) if service_center is not None: persisted_pk(service_center, ServiceCenter) # Center creation uses the same role path through a scoped context helper. return _create_center_location(actor=actor, company=company, service_center=service_center, code=code, name=name, location_type=location_type) with inventory_context(actor=actor, company=company, permission="inventory.manage_inventory") as (_, company, _, _): row = InventoryLocation(company=company, code=text_value(code, required=True, maximum=64).upper(), name=text_value(name, required=True, maximum=200), location_type=location_type) row._persist() return row @transaction.atomic def _create_center_location(*, actor, company, service_center, code, name, location_type): from django.contrib.auth import get_user_model from apps.access.locks import share_rows from apps.access.models import Role, UserRoleAssignment from apps.organization.assignments import UserOrganizationAssignment user_id = persisted_pk(actor, get_user_model()) share_rows(get_user_model(), [user_id]) actor = get_user_model().objects.get(pk=user_id) share_rows(Company, [company.pk]) company = Company.objects.get(pk=company.pk) paths = UserOrganizationAssignment.objects.filter(user=actor, company=company) share_rows(UserOrganizationAssignment, paths.values_list("pk", flat=True)) share_rows(Role, UserRoleAssignment.objects.filter(user=actor, organization_assignment__in=paths).values_list("role_id", flat=True)) center = ServiceCenter.objects.select_related("region").get(pk=service_center.pk) require_permission(user=actor, permission="inventory.manage_inventory", target=center) if not (actor.is_active and company.is_active and center.is_active and center.region.is_active and center.company_id == company.pk == center.region.company_id): raise ValidationError("An active center in this company is required.") row = InventoryLocation(company=company, service_center=center, code=text_value(code, required=True, maximum=64).upper(), name=text_value(name, required=True, maximum=200), location_type=location_type) row._persist() return row def update_location(*, actor, location, name, expected_revision): with inventory_context(actor=actor, company=location.company, permission="inventory.manage_inventory", locations=[location], exclusive_locations=True) as (_, _, locations, _): row = locations[location.pk] check_revision(row, expected_revision) if row.location_type in ("TRANSIT", "CUSTODY"): raise ValidationError("Transit locations are managed by transfer workflows.") row.name = text_value(name, required=True, maximum=200) row._persist() return row def _location_lifecycle(*, actor, location, active, expected_revision): from django.db.models import Sum with inventory_context(actor=actor, company=location.company, permission="inventory.manage_inventory", locations=[location], exclusive_locations=True) as (_, _, locations, _): row = locations[location.pk] check_revision(row, expected_revision) if row.location_type in ("TRANSIT", "CUSTODY"): raise ValidationError("Transit locations are managed by transfer workflows.") from .models import StockTransfer from .models import DefectiveRecovery if not active and DefectiveRecovery.objects.filter(location=row).exists(): raise ValidationError("A location holding recovered customer components cannot be deactivated.") from .models import StockCount if not active and StockCount.objects.filter(location=row, status="COUNTING").exists(): raise ValidationError("An active physical count prevents location deactivation.") from django.db.models import Q if not active and StockTransfer.objects.filter(status="DISPATCHED").filter(Q(source=row) | Q(destination=row)).exists(): raise ValidationError("A dispatched transfer still depends on this location.") if not active and row.ledger_entries.values("spare_part_id").annotate(balance=Sum("quantity_delta")).exclude(balance=0).exists(): raise ValidationError("A location holding stock cannot be deactivated.") row.is_active = active row._persist() return row def deactivate_location(*, actor, location, expected_revision=None): return _location_lifecycle(actor=actor, location=location, active=False, expected_revision=expected_revision) def reactivate_location(*, actor, location, expected_revision=None): return _location_lifecycle(actor=actor, location=location, active=True, expected_revision=expected_revision) def register_serialized_unit(*, actor, company, spare_part, identifier): with inventory_context(actor=actor, company=company, parts=[spare_part], permission="inventory.receive_stock") as (actor, company, _, parts): part = parts[spare_part.pk] return _register_unit(actor=actor, company=company, part=part, identifier=identifier) def normalized_identifier(identifier): identifier = text_value(identifier, required=True, maximum=128) if any(not char.isprintable() for char in identifier): raise ValidationError("Identifiers must not contain control characters.") return identifier def _register_unit(*, actor, company, part, identifier): # Called only with part/dependency locks and caller authorization established. if part.serialization_policy == "NOT_SERIALIZED": raise ValidationError("This part does not permit serialized inventory.") row = SerializedStockUnit(company=company, spare_part=part, identifier=normalized_identifier(identifier), registered_by=actor) row._persist() return row def _post(*, actor, company, spare_part, destination, quantity, reference, idempotency_key, source=None, units=(), note="", kind=None, count=None): """Internal: dependencies, endpoint and stock-position locks already held. This one writer is reused by document workflows. Never call it as an HTTP API. """ quantity = quantity_value(quantity) if not isinstance(idempotency_key, uuid.UUID): raise ValidationError("Supply an explicit UUID command key.") kind = kind or ("MOVE" if source else "RECEIPT") if (destination is None and (kind not in ("CONSUME", "ADJUST_OUT") or source is None)) or (kind in ("CONSUME", "ADJUST_OUT") and destination is not None): raise ValidationError("Consumption requires an outward-only movement.") endpoints = ([destination] if destination else []) + ([source] if source else []) from .control_services import require_position_open for endpoint in endpoints: require_position_open(endpoint, spare_part, count=count) if any(not row.is_active or row.company_id != company.pk for row in endpoints): raise ValidationError("Active same-company locations are required.") if source and destination and source.pk == destination.pk: raise ValidationError("Source and destination must differ.") if source and destination and source.location_type in ("QUARANTINE", "DEFECTIVE") and destination.location_type in USABLE_TYPES: raise ValidationError("Non-serviceable stock cannot be silently made usable.") unit_ids = [persisted_pk(obj, SerializedStockUnit) for obj in units] if len(set(unit_ids)) != len(unit_ids) or len(unit_ids) > quantity: raise ValidationError("Supply distinct units within the movement quantity.") if spare_part.serialization_policy == "REQUIRED_SERIAL" and len(unit_ids) != quantity: raise ValidationError("Every unit requires its own registered identifier.") if spare_part.serialization_policy == "NOT_SERIALIZED" and unit_ids: raise ValidationError("Serialized units are not permitted for this part.") current_units = list(SerializedStockUnit.objects.filter(pk__in=unit_ids).select_for_update().order_by("pk")) if len(current_units) != len(unit_ids): raise ValidationError("A serialized unit no longer exists.") for unit in current_units: if unit.company_id != company.pk or unit.spare_part_id != spare_part.pk: raise ValidationError("Unit company/part does not match this movement.") source_state = {"TRANSIT": "IN_TRANSIT", "CUSTODY": "IN_CUSTODY"}.get(source.location_type if source else None, "IN_STOCK") if (source and (unit.current_location_id != source.pk or unit.state != source_state)) or ( not source and (unit.current_location_id is not None or unit.state not in (("REGISTERED", "REMOVED") if kind == "ADJUST_IN" else ("REGISTERED",)))): raise ValidationError("Unit is not in the required inventory state/location.") if source: from .request_services import require_unreserved require_unreserved(location=source, part=spare_part, quantity=quantity, units=current_units) on_hand = _on_hand(source, spare_part) serialized = SerializedStockUnit.objects.filter(current_location=source, spare_part=spare_part).count() if on_hand < quantity or on_hand - serialized < quantity - len(unit_ids): raise ValidationError("Insufficient stock in the selected serialized/non-serialized bucket.") movement = StockMovement(company=company, spare_part=spare_part, kind=kind, quantity=quantity, source=source, destination=destination, actor=actor, reference=text_value(reference, required=True, maximum=128), note=text_value(note), idempotency_key=idempotency_key) movement._persist() for location, delta in ([(destination, quantity)] if destination else []) + ([(source, -quantity)] if source else []): StockLedgerEntry(movement=movement, company=company, spare_part=spare_part, location=location, quantity_delta=delta)._persist() for unit in current_units: StockMovementUnit(movement=movement, unit=unit)._persist() unit.current_location = destination unit.current_movement = movement unit.state = ({"TRANSIT": "IN_TRANSIT", "CUSTODY": "IN_CUSTODY"}.get(destination.location_type, "IN_STOCK") if destination else ("REMOVED" if kind == "ADJUST_OUT" else "CONSUMED")) unit._persist() return movement def receive_stock(*, actor, destination, spare_part, quantity, reference, idempotency_key, units=(), note="", expected_revision=None): with inventory_context(actor=actor, company=destination.company, permission="inventory.receive_stock", locations=[destination], parts=[spare_part]) as (actor, company, locations, parts): destination = locations[destination.pk] if destination.location_type in ("TRANSIT", "CUSTODY"): raise ValidationError("Use the transfer workflow for transit stock.") check_revision(destination, expected_revision) part = parts[spare_part.pk] lock_positions([destination], [part]) return _post(actor=actor, company=company, spare_part=part, destination=destination, quantity=quantity, reference=reference, idempotency_key=idempotency_key, units=units, note=note) def move_stock(*, actor, source, destination, spare_part, quantity, reference, idempotency_key, units=(), note="", expected_source_revision=None, expected_destination_revision=None): with inventory_context(actor=actor, company=source.company, permission="inventory.transfer_stock", locations=[source, destination], parts=[spare_part]) as (actor, company, locations, parts): source, destination = locations[source.pk], locations[destination.pk] if any(row.location_type in ("TRANSIT", "CUSTODY") for row in (source, destination)): raise ValidationError("Use the transfer workflow for transit stock.") check_revision(source, expected_source_revision) check_revision(destination, expected_destination_revision) part = parts[spare_part.pk] lock_positions([source, destination], [part]) return _post(actor=actor, company=company, spare_part=part, source=source, destination=destination, quantity=quantity, reference=reference, idempotency_key=idempotency_key, units=units, note=note)