add ! External Beneficiary Reference field

Author: vitali-yanushchyk-valor

src/country_workspace/contrib/hope/apps.py

@@ -3,6 +3,7 @@
 from .geo import Admin1Choice, Admin2Choice, Admin3Choice, Admin4Choice, CountryChoice
 from .lookups import FinancialInstitutionChoice
 from .phone_numbers import PhoneNumberField
+from .beneficiary_reference import BeneficiaryReferenceModelChoice
 
 
 class Config(AppConfig):
@@ -23,6 +24,7 @@ def ready(self) -> None:
         field_registry.register(Admin4Choice)
         field_registry.register(FinancialInstitutionChoice)
         field_registry.register(PhoneNumberField)
+        field_registry.register(BeneficiaryReferenceModelChoice)
 
         from country_workspace.contrib.hope.validators import FullHouseholdValidator
         from country_workspace.validators.registry import beneficiary_validator_registry

src/country_workspace/contrib/hope/beneficiary_reference.py

@@ -0,0 +1,112 @@
+from typing import Any, TYPE_CHECKING
+from contextlib import suppress
+from django.core.exceptions import ValidationError
+from django.db.models import QuerySet
+from django.forms import ModelChoiceField
+from django.urls import resolve
+from django_select2.forms import ModelSelect2Widget
+from country_workspace.state import state
+
+if TYPE_CHECKING:
+    from country_workspace.models import Individual
+
+
+class BeneficiarySelect2Widget(ModelSelect2Widget):
+    search_fields = ["name__icontains"]
+
+    def __init__(self, batch_id: int | None = None, *args: Any, **kwargs: Any) -> None:
+        self.batch_id = batch_id
+        kwargs.setdefault(
+            "attrs",
+            {
+                "data-minimum-input-length": 0,
+                "class": "form-control",
+            },
+        )
+        super().__init__(*args, **kwargs)
+
+    def get_queryset(self) -> QuerySet["Individual"]:
+        return _get_individuals_queryset(self.batch_id)
+
+
+class BeneficiaryReferenceModelChoice(ModelChoiceField):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        batch_id = _get_batch_id_from_request()
+        queryset = _get_individuals_queryset(batch_id)
+
+        kwargs.setdefault("queryset", queryset)
+        if batch_id:
+            kwargs.setdefault("widget", BeneficiarySelect2Widget(batch_id=batch_id, model=queryset.model))
+
+        super().__init__(*args, **kwargs)
+
+    def to_python(self, value: Any) -> str | None:
+        if value in self.empty_values:
+            return None
+
+        match value:
+            case obj if hasattr(obj, "name"):
+                return obj.name
+            case int() | str() if str(value).isdigit():
+                with suppress(self.queryset.model.DoesNotExist):
+                    return self.queryset.get(pk=int(value)).name
+            case str():
+                with suppress(self.queryset.model.DoesNotExist):
+                    return self.queryset.get(flex_fields__individual_id=value).name
+                with suppress(self.queryset.model.DoesNotExist):
+                    return self.queryset.get(name=value).name
+            case _:
+                pass
+
+        raise ValidationError(self.error_messages["invalid_choice"], code="invalid_choice")
+
+    def prepare_value(self, value: Any) -> int | None:
+        if value in self.empty_values:
+            return None
+
+        match value:
+            case obj if hasattr(obj, "pk"):
+                return obj.pk
+            case str():
+                with suppress(self.queryset.model.DoesNotExist):
+                    return self.queryset.get(flex_fields__individual_id=value).pk
+                with suppress(self.queryset.model.DoesNotExist):
+                    return self.queryset.get(name=value).pk
+            case _:
+                pass
+
+        return None
+
+
+def _get_batch_id_from_request() -> int | None:
+    """Extract batch ID from current request context."""
+    if not state.request:
+        return None
+
+    resolved = resolve(state.request.path)
+    if not (resolved and "object_id" in resolved.kwargs):
+        return None
+    if resolved.view_name != "workspace:workspaces_countryhousehold_change":
+        return None
+
+    try:
+        from django.apps import apps
+
+        Household = apps.get_model("country_workspace", "Household")
+        household = Household.objects.select_related("batch").get(pk=resolved.kwargs["object_id"])
+    except (Household.DoesNotExist, ValueError, KeyError):
+        return None
+    else:
+        return household.batch.pk
+
+
+def _get_individuals_queryset(batch_id: int | None) -> QuerySet["Individual"]:
+    """Get filtered Individual queryset for given batch_id."""
+    from django.apps import apps
+
+    Individual = apps.get_model("country_workspace", "Individual")
+
+    if not batch_id:
+        return Individual.objects.none()
+
+    return Individual.objects.filter(batch_id=batch_id, removed=False)

src/country_workspace/migrations/0022_individual_unique_name_per_batch_individual.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.2.3 on 2025-08-01 13:16
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("country_workspace", "0021_dataserializer_program_serializer"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="individual",
+            constraint=models.UniqueConstraint(
+                fields=("batch", "name"),
+                name="unique_name_per_batch_individual",
+                violation_error_message="Name must be unique per batch.",
+            ),
+        ),
+    ]

src/country_workspace/models/base.py

@@ -76,6 +76,13 @@ class Meta:
             ("export_beneficiary", "Can Export Beneficiary Records"),
             ("push_beneficiary_to_hope", "Can Push Beneficiary Records To HOPE core"),
         )
+        constraints = [
+            models.UniqueConstraint(
+                fields=["batch", "name"],
+                name="unique_name_per_batch_%(class)s",
+                violation_error_message=_("Name must be unique per batch."),
+            ),
+        ]
 
     def __str__(self) -> str:
         return self.name or "%s %s" % (self._meta.verbose_name, self.id)

src/country_workspace/versioning/scripts/0016_add_beneficiary_reference.py

@@ -0,0 +1,90 @@
+# Generated by HCW 0.1.0 on 2025 07 29 07:39:07
+from typing import Any
+from packaging.version import Version
+from concurrency.utils import fqn
+from contextlib import suppress
+from django import forms
+from django.db import transaction
+from django.utils.text import slugify
+
+from hope_flex_fields.models import FieldDefinition, Fieldset
+from hope_flex_fields.utils import get_kwargs_from_field_class, get_common_attrs
+from country_workspace.contrib.hope.beneficiary_reference import BeneficiaryReferenceModelChoice
+from country_workspace.contrib.hope.constants import HOUSEHOLD_FIELDSET_NAME, INDIVIDUAL_FIELDSET_NAME
+
+_script_for_version = Version("0.1.0")
+
+
+attrs_default = lambda cls: get_kwargs_from_field_class(cls, get_common_attrs())
+
+FIELDS: dict[str, dict[str, Any]] = {
+    "alternate_collector_id": {
+        "name": "Alternate Collector Reference ID",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "help_text": "A unique beneficiary reference ID for an alternate collector.",
+            },
+        },
+    },
+    "primary_collector_id": {
+        "name": "Primary Collector Reference ID",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "help_text": "A unique beneficiary reference ID for a primary collector.",
+            },
+        },
+    },
+    "individual_id": {
+        "name": "Individual ID",
+        "defaults": {
+            "field_type": fqn(forms.CharField),
+            "attrs": {
+                **attrs_default(forms.CharField),
+                "help_text": "A unique individual ID.",
+            },
+        },
+    },
+}
+
+
+FIELDSETS = {
+    HOUSEHOLD_FIELDSET_NAME: ("alternate_collector_id", "primary_collector_id"),
+    INDIVIDUAL_FIELDSET_NAME: ("individual_id",),
+}
+
+
+def forward() -> None:
+    with transaction.atomic():
+        for fieldset_name, field_names in FIELDSETS.items():
+            fs, __ = Fieldset.objects.get_or_create(name=fieldset_name)
+            for field_name in field_names:
+                field_def = FIELDS[field_name]
+                field_def["defaults"]["slug"] = slugify(field_def["name"])
+                fd, __ = FieldDefinition.objects.update_or_create(
+                    name=field_def["name"], defaults=field_def["defaults"]
+                )
+                fs.fields.update_or_create(name=field_name, defaults={"definition": fd})
+
+
+def backward() -> None:
+    with transaction.atomic():
+        for fieldset_name, field_names in FIELDSETS.items():
+            try:
+                fs = Fieldset.objects.get(name=fieldset_name)
+            except Fieldset.DoesNotExist:
+                continue
+
+            for field_name in field_names:
+                fs.fields.filter(name=field_name).delete()
+                with suppress(FieldDefinition.DoesNotExist):
+                    fd = FieldDefinition.objects.get(name=FIELDS[field_name]["name"])
+                    fd.delete()
+
+
+class Scripts:
+    requires = []
+    operations = [(forward, backward)]