import uuid from django.core.exceptions import ValidationError from django.core.validators import RegexValidator from django.db import models, router, transaction from .model_base import TimeStampedModel CODE_PATTERN = r"^[A-Z0-9][A-Z0-9_-]*$" class OrganizationRecord(TimeStampedModel): """Fields and write validation shared by this domain's master records.""" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) code = models.CharField(max_length=32, validators=[RegexValidator( CODE_PATTERN, "Use letters, digits, hyphens or underscores; start with a letter or digit." )]) name = models.CharField(max_length=200) is_active = models.BooleanField(default=True) class Meta: abstract = True ordering = ["code"] constraints = [models.CheckConstraint( condition=models.Q(code__regex=CODE_PATTERN), name="%(app_label)s_%(class)s_code_format", )] def clean_fields(self, exclude=None): self.code = self.code.strip().upper() if isinstance(self.code, str) else self.code super().clean_fields(exclude=exclude) def save(self, *, force_insert=False, force_update=False, using=None, update_fields=None): if update_fields is not None: update_fields = set(update_fields) if not update_fields: return update_fields.add("updated_at") using = using or router.db_for_write(type(self), instance=self) with transaction.atomic(using=using): # One company row coordinates hierarchy writes, including new children. company_id = self.pk if isinstance(self, Company) else self.company_id Company.objects.using(using).select_for_update().filter(pk=company_id).first() self.full_clean() if update_fields is not None: persisted = type(self).objects.using(using).filter(pk=self.pk).first() if persisted is not None: # Validate what will actually be persisted, not just the in-memory object. for field in self._meta.concrete_fields: if field.name in update_fields or field.attname in update_fields: setattr(persisted, field.attname, getattr(self, field.attname)) persisted.full_clean() return super().save( force_insert=force_insert, force_update=force_update, using=using, update_fields=update_fields, ) def clean(self): super().clean() if not self.is_active and self.pk and self.user_assignments.filter(is_active=True).exists(): raise ValidationError({"is_active": "Active user assignments exist. Use the deactivation service/action."}) if not isinstance(self, Company) and self.company_id: original_company = type(self).objects.filter(pk=self.pk).values_list("company_id", flat=True).first() if original_company is not None and original_company != self.company_id: raise ValidationError({"company": "Company ownership cannot change after creation."}) if self.is_active and Company.objects.filter(pk=self.company_id, is_active=False).exists(): raise ValidationError({"is_active": "An active record requires an active company."}) def __str__(self): return f"{self.code} ? {self.name}" class Company(OrganizationRecord): legal_name = models.CharField(max_length=255, blank=True) class Meta(OrganizationRecord.Meta): abstract = False verbose_name_plural = "companies" constraints = OrganizationRecord.Meta.constraints + [ models.UniqueConstraint(fields=["code"], name="org_company_code_uniq"), ] def clean(self): super().clean() if not self.is_active and self.pk: if any(relation.filter(is_active=True).exists() for relation in [self.regions, self.service_centers, self.departments]): raise ValidationError({"is_active": "Active children exist. Use the company deactivation service/action."}) class Region(OrganizationRecord): company = models.ForeignKey(Company, on_delete=models.PROTECT, related_name="regions") class Meta(OrganizationRecord.Meta): abstract = False constraints = OrganizationRecord.Meta.constraints + [ models.UniqueConstraint(fields=["company", "code"], name="org_region_company_code_uniq"), ] def clean(self): super().clean() if not self.is_active and self.pk and self.service_centers.filter(is_active=True).exists(): raise ValidationError({"is_active": "Active centers exist. Use the region deactivation service/action."}) class ServiceCenter(OrganizationRecord): class CenterType(models.TextChoices): OWN = "OWN", "Own Service Center" AUTHORIZED = "AUTHORIZED", "Authorized Service Center" PARTNER = "PARTNER", "Partner Service Center" company = models.ForeignKey(Company, on_delete=models.PROTECT, related_name="service_centers") region = models.ForeignKey(Region, on_delete=models.PROTECT, related_name="service_centers") center_type = models.CharField(max_length=10, choices=CenterType.choices) phone = models.CharField(max_length=32, blank=True) email = models.EmailField(blank=True) address = models.TextField(blank=True) class Meta(OrganizationRecord.Meta): abstract = False constraints = OrganizationRecord.Meta.constraints + [ models.UniqueConstraint(fields=["company", "code"], name="org_center_company_code_uniq"), models.CheckConstraint(condition=models.Q(center_type__in=["OWN", "AUTHORIZED", "PARTNER"]), name="org_center_type_valid"), ] def clean(self): super().clean() previous_region = type(self).objects.filter(pk=self.pk).values_list("region_id", flat=True).first() if previous_region is not None and previous_region != self.region_id and self.user_assignments.exists(): raise ValidationError({"region": "Cannot move a center referenced by assignments, including history."}) if self.region_id and self.company_id: # Read current state; relation caches may predate lifecycle changes. region = Region.objects.filter(pk=self.region_id).values("company_id", "is_active").first() if region is not None: if region["company_id"] != self.company_id: raise ValidationError({"region": "Region must belong to the selected company."}) if self.is_active and not region["is_active"]: raise ValidationError({"is_active": "An active service center requires an active region."}) class Department(OrganizationRecord): company = models.ForeignKey(Company, on_delete=models.PROTECT, related_name="departments") class Meta(OrganizationRecord.Meta): abstract = False constraints = OrganizationRecord.Meta.constraints + [ models.UniqueConstraint(fields=["company", "code"], name="org_department_company_code_uniq"), ] # Register the assignment model without coupling accounts to organization. from .assignments import UserOrganizationAssignment # noqa: E402, F401