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,101 @@
+from typing import Any, TYPE_CHECKING
+from django.forms import ModelChoiceField
+from django_select2.forms import ModelSelect2Widget
+from django.db.models import QuerySet
+from django.core.exceptions import ValidationError
+from django.urls import resolve
+from country_workspace.state import state
+
+if TYPE_CHECKING:
+    from country_workspace.models import Individual
+
+
+def get_household_id_from_request() -> int | None:
+    if not state.request:
+        return None
+
+    if (resolved := resolve(state.request.path)) and "object_id" in resolved.kwargs:
+        return int(resolved.kwargs["object_id"])
+    return None
+
+
+class BeneficiarySelect2Widget(ModelSelect2Widget):
+    search_fields = ["name__icontains"]
+
+    def __init__(
+        self, program_id: int | None = None, household_id: int | None = None, *args: Any, **kwargs: Any
+    ) -> None:
+        self.program_id = program_id
+        self.household_id = household_id
+        kwargs.setdefault(
+            "attrs",
+            {
+                "data-minimum-input-length": 0,
+                "class": "form-control",
+            },
+        )
+        super().__init__(*args, **kwargs)
+
+    def get_queryset(self) -> QuerySet["Individual"]:
+        from django.apps import apps
+
+        Individual = apps.get_model("country_workspace", "Individual")
+
+        if not self.program_id:
+            return Individual.objects.none()
+
+        try:
+            Program = apps.get_model("country_workspace", "Program")
+            program = Program.objects.get(id=self.program_id)
+            queryset = program.individuals.filter(removed=False)
+            if self.household_id:
+                queryset = queryset.filter(household=self.household_id)
+        except Program.DoesNotExist:
+            return Individual.objects.none()
+        else:
+            return queryset
+
+
+class BeneficiaryReferenceModelChoice(ModelChoiceField):
+    def __init__(self, filter_by_household: bool = False, *args: Any, **kwargs: Any) -> None:
+        program = state.program
+        household_id = get_household_id_from_request() if filter_by_household else None
+
+        queryset = program.individuals.filter(removed=False)
+        if household_id:
+            queryset = queryset.filter(household=household_id)
+
+        kwargs.setdefault("queryset", queryset)
+        kwargs.setdefault(
+            "widget", BeneficiarySelect2Widget(program_id=program.id, household_id=household_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 int() | str() if str(value).isdigit():
+                try:
+                    individual = self.queryset.get(pk=int(value))
+                except self.queryset.model.DoesNotExist:
+                    raise ValidationError(self.error_messages["invalid_choice"], code="invalid_choice")
+                return individual.name
+
+            case str() if self.queryset.filter(name=value).exists():
+                return value
+
+            case _:
+                raise ValidationError(self.error_messages["invalid_choice"], code="invalid_choice")
+
+    def prepare_value(self, value: Any) -> int | None:
+        if isinstance(value, str) and value:
+            try:
+                individual = self.queryset.get(name=value)
+            except self.queryset.model.DoesNotExist:
+                return None
+            else:
+                return individual.pk
+        return getattr(value, "pk", value)

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

@@ -0,0 +1,72 @@
+# 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.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
+
+_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": {
+        "name": "Alternate Collector Reference",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "filter_by_household": False,
+                "help_text": "A unique reference for an alternate collector.",
+            },
+        },
+    },
+    "primary_collector": {
+        "name": "Primary Collector Reference",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "filter_by_household": True,
+                "help_text": "A unique reference for a primary collector.",
+            },
+        },
+    },
+}
+
+
+def forward() -> None:
+    with transaction.atomic():
+        fs, __ = Fieldset.objects.get_or_create(name=HOUSEHOLD_FIELDSET_NAME)
+        for field_name, field_config in FIELDS.items():
+            field_config["defaults"]["slug"] = slugify(field_config["name"])
+            fd, __ = FieldDefinition.objects.update_or_create(
+                name=field_config["name"], defaults=field_config["defaults"]
+            )
+            fs.fields.update_or_create(name=field_name, defaults={"definition": fd})
+
+
+def backward() -> None:
+    with transaction.atomic():
+        try:
+            fs = Fieldset.objects.get(name=HOUSEHOLD_FIELDSET_NAME)
+        except Fieldset.DoesNotExist:
+            return
+
+        for field_name, field_config in FIELDS.items():
+            fs.fields.filter(name=field_name).delete()
+            with suppress(FieldDefinition.DoesNotExist):
+                fd = FieldDefinition.objects.get(name=field_config["name"])
+                fd.delete()
+
+
+class Scripts:
+    requires = []
+    operations = [(forward, backward)]