AB#242087: Add base64 image field

pyproject.toml

@@ -52,6 +52,7 @@ dependencies = [
     "unicef-security>=1.5.1",
     "django-pghistory>=3.5.4",
     "django-anymail[mailjet]>=13.0",
+    "pillow>=11.2.1",
 ]
 [project.scripts]
 celery-monitor = "country_workspace.__monitor__:run"

src/country_workspace/utils/flex_fields.py

@@ -1,9 +1,15 @@
+from base64 import b64encode
 import hashlib
 import json
-from typing import TYPE_CHECKING, Generator
+from typing import TYPE_CHECKING, Generator, Literal
+
+from django import forms
+from django.core.files.uploadedfile import UploadedFile
 
 from hope_flex_fields.models import DataChecker
 
+from country_workspace.contrib.kobo.api.data.helpers import VALUE_FORMAT
+
 if TYPE_CHECKING:
     from country_workspace.models.base import Validable
 
@@ -21,3 +27,23 @@ def get_obj_checksum(obj: "Validable") -> str:
     if obj.flex_files:
         h.update(obj.flex_files[:8192])  # is this enough ?
     return h.hexdigest()
+
+
+class Base64ImageInput(forms.ClearableFileInput):
+    template_name = "workspace/base64_image_widget.html"
+
+    def is_initial(self, value: str | None) -> bool:
+        # we need to override this as base method looks for url
+        return bool(value)
+
+
+class Base64ImageField(forms.ImageField):
+    widget = Base64ImageInput
+
+    def clean(self, data: UploadedFile | Literal[False], initial: str | None = None) -> str | None:
+        if cleaned_data := super().clean(data, initial):
+            content = b64encode(cleaned_data.read()).decode()
+            return VALUE_FORMAT.format(mimetype=data.content_type, content=content)
+
+        # if we return cleaned_data here, False will be stored, so we return None explicitly
+        return None

src/country_workspace/versioning/scripts/0004_add_base64_image_field.py

@@ -0,0 +1,31 @@
+# Generated by HCW 0.1.0 on 2025 04 15 09:25:45
+from concurrency.utils import fqn
+from hope_flex_fields.models import FieldDefinition
+from hope_flex_fields.registry import field_registry
+from hope_flex_fields.utils import get_kwargs_from_field_class, get_common_attrs
+from packaging.version import Version
+
+from country_workspace.utils.flex_fields import Base64ImageField
+
+_script_for_version = Version("0.1.0")
+
+
+def forward() -> None:
+    field_registry.register(Base64ImageField)
+    FieldDefinition.objects.get_or_create(
+        name=Base64ImageField.__name__,
+        field_type=fqn(Base64ImageField),
+        defaults={"attrs": get_kwargs_from_field_class(Base64ImageField, get_common_attrs())},
+    )
+
+
+def backward() -> None:
+    FieldDefinition.objects.get(
+        name=Base64ImageField.__name__,
+        field_type=fqn(Base64ImageField),
+    ).delete()
+
+
+class Scripts:
+    requires = []
+    operations = [(forward, backward)]

src/country_workspace/workspaces/admin/hh_ind.py

@@ -22,7 +22,7 @@
 from ..models import CountryHousehold, CountryIndividual
 from .cleaners import actions
 from .cleaners.validate import validate_program
-
+from ...utils.flex_fields import Base64ImageField
 
 if TYPE_CHECKING:
     from hope_flex_fields.forms import FlexForm
@@ -231,7 +231,7 @@ def _changeform_view(
             initials = {}
         if request.method == "POST":
             if obj:
-                form: "FlexForm" = form_class(request.POST, prefix="flex_field", initial=initials)
+                form: "FlexForm" = form_class(request.POST, request.FILES, prefix="flex_field", initial=initials)
                 form_valid = form.is_valid()
                 if form_valid or "_save_invalid" in request.POST:
                     obj.flex_fields = form.cleaned_data
@@ -248,6 +248,7 @@ def _changeform_view(
         context["show_save_invalid"] = True
         context["checker_form"] = form
         context["has_change_permission"] = self.has_change_permission(request)
+        context["has_file_field"] = any(isinstance(field, Base64ImageField) for field in form.fields.values())
 
         return TemplateResponse(request, self.change_form_template, context)
 

src/country_workspace/workspaces/templates/workspace/base64_image_widget.html

@@ -0,0 +1,12 @@
+{% if widget.is_initial %}
+    {{ widget.initial_text }}:
+    <img src="{{ widget.value }}" style="max-height: 100px; width: auto">
+    {% if not widget.required %}
+        <input type="checkbox" name="{{ widget.checkbox_name }}" id="{{ widget.checkbox_id }}"
+                {% if widget.attrs.disabled %} disabled{% endif %}
+                {% if widget.attrs.checked %} checked{% endif %}>
+        <label for="{{ widget.checkbox_id }}">{{ widget.clear_checkbox_label }}</label>
+    {% endif %}<br>
+    {{ widget.input_text }}:
+{% endif %}
+<input type="{{ widget.type }}" name="{{ widget.name }}"{% include "django/forms/widgets/attrs.html" %}>

tests/utils/test_utils_fields.py

@@ -1,12 +1,17 @@
+from unittest.mock import Mock
+
 import pytest
-from pytest_mock import MockFixture
+from django.core.files.uploadedfile import SimpleUploadedFile
+from pytest_mock import MockerFixture
 
+from country_workspace.contrib.kobo.api.data.helpers import VALUE_FORMAT
 from country_workspace.utils.fields import (
     clean_field_name,
     TO_REMOVE_VALUES,
     clean_field_names,
     map_fields,
 )
+from country_workspace.utils.flex_fields import Base64ImageInput, Base64ImageField
 
 
 @pytest.mark.parametrize(
@@ -21,7 +26,7 @@ def test_clean_field_name(input_value, expected_output):
     assert clean_field_name(input_value) == expected_output
 
 
-def test_clean_field_names(mocker: MockFixture) -> None:
+def test_clean_field_names(mocker: MockerFixture) -> None:
     clean_field_name_mock = mocker.patch("country_workspace.utils.fields.clean_field_name")
 
     cleaned = clean_field_names({(key := "foo"): "bar"})
@@ -42,3 +47,28 @@ def test_clean_field_names(mocker: MockFixture) -> None:
 def test_map_fields(input_fields, expected_output):
     result = map_fields(input_fields)
     assert result == expected_output
+
+
+@pytest.mark.parametrize("value", [None, "", "test"])
+def test_base64_image_input(value: str | None) -> None:
+    input_ = Base64ImageInput()
+    assert input_.is_initial(value) == bool(value)
+
+
+def test_base64_image_field_file_was_cleared(mocker: MockerFixture) -> None:
+    super_clean_mock = mocker.patch("country_workspace.utils.flex_fields.forms.ImageField.clean")
+    super_clean_mock.return_value = False
+    instance = Mock(spec=Base64ImageField)
+
+    assert Base64ImageField.clean(instance, False) is None
+
+
+def test_base64_image_field_content_is_encoded(mocker: MockerFixture) -> None:
+    super_clean_mock = mocker.patch("country_workspace.utils.flex_fields.forms.ImageField.clean")
+    b64encode_mock = mocker.patch("country_workspace.utils.flex_fields.b64encode")
+    b64encode_mock.return_value.decode.return_value = (data := "decoded")
+    file = SimpleUploadedFile("test.txt", b"test", content_type=(content_type := "text/plain"))
+    super_clean_mock.return_value = file
+    instance = Mock(spec=Base64ImageField)
+
+    assert Base64ImageField.clean(instance, file) == VALUE_FORMAT.format(mimetype=content_type, content=data)