Implement django forms

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+  // Use IntelliSense to learn about possible attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Python Debugger: Django",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${workspaceFolder}/manage.py",
+      "args": ["runserver"],
+      "django": true
+    }
+  ]
+}

home/forms/search.py

@@ -0,0 +1,76 @@
+from copy import deepcopy
+from django import forms
+from urllib.parse import urlencode
+
+# from home.helper import get_domain_list
+
+
+def get_domain_choices():
+    """Make API call to obtain domain choices"""
+    # TODO: pull in the domains from the catalogue client
+    # facets = client.search_facets()
+    # domain_list = facets.options("domains")
+    return [
+        ("urn:li:domain:HMCTS", "HMCTS"),
+        ("urn:li:domain:HMPPS", "HMPPS"),
+        ("urn:li:domain:OPG", "OPG"),
+        ("urn:li:domain:HQ", "HQ"),
+    ]
+
+
+def get_sort_choices():
+    return [
+        ("relevance", "Relevance"),
+        ("ascending", "Ascending"),
+        ("descending", "Descending"),
+    ]
+
+
+class SearchForm(forms.Form):
+    """Django form to represent data product search page inputs"""
+
+    query = forms.CharField(
+        max_length=100,
+        strip=False,
+        required=False,
+        widget=forms.TextInput(attrs={"class": "govuk-input search-input"}),
+    )
+    domains = forms.MultipleChoiceField(
+        choices=get_domain_choices,
+        required=False,
+        widget=forms.CheckboxSelectMultiple(
+            attrs={"class": "govuk-checkboxes__input", "form": "searchform"}
+        ),
+    )
+    sort = forms.ChoiceField(
+        choices=get_sort_choices,
+        widget=forms.RadioSelect(
+            attrs={
+                "class": "govuk-radios__input",
+                "form": "searchform",
+                "onchange": "document.getElementById('searchform').submit();",
+            }
+        ),
+        required=False,
+    )
+    clear_filter = forms.BooleanField(initial=False, required=False)
+    clear_label = forms.BooleanField(initial=False, required=False)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.initial["sort"] = "relevance"
+
+    def clean_query(self):
+        """Example clean method to apply custom validation to input fields"""
+        return str(self.cleaned_data["query"]).capitalize()
+
+    def encode_without_filter(self, filter_to_remove):
+        """Preformat hrefs to drop individual filters"""
+        # Deepcopy the cleaned data dict to avoid modifying it inplace
+        query_params = deepcopy(self.cleaned_data)
+
+        query_params["domains"].remove(filter_to_remove)
+        if len(query_params["domains"]) == 0:
+            query_params.pop("domains")
+
+        return f"?{urlencode(query_params, doseq=True)}"

home/service/base.py

@@ -0,0 +1,11 @@
+from data_platform_catalogue.client import BaseCatalogueClient
+from data_platform_catalogue.client.datahub import DataHubCatalogueClient
+from django.conf import settings
+
+
+class GenericService:
+    @staticmethod
+    def _get_catalogue_client() -> BaseCatalogueClient:
+        return DataHubCatalogueClient(
+            jwt_token=settings.CATALOGUE_TOKEN, api_url=settings.CATALOGUE_URL
+        )

home/service/details.py

@@ -0,0 +1,33 @@
+from data_platform_catalogue.search_types import MultiSelectFilter, ResultType
+from django.core.exceptions import ObjectDoesNotExist
+
+from .base import GenericService
+
+
+class DetailsService(GenericService):
+    def __init__(self, urn: str):
+        self.urn = urn
+        self.client = self._get_catalogue_client()
+
+        filter_value = [MultiSelectFilter("urn", [urn])]
+        search_results = self.client.search(
+            query="", page=None, filters=filter_value)
+
+        if not search_results.page_results:
+            raise ObjectDoesNotExist(urn)
+
+        self.result = search_results.page_results[0]
+        self.context = self._get_context()
+
+    def _get_context(self):
+        context = {
+            "result": self.result,
+            "result_type": (
+                "Data product"
+                if self.result.result_type == ResultType.DATA_PRODUCT
+                else "Table"
+            ),
+            "page_title": f"{self.result.name} - Data catalogue",
+        }
+
+        return context

home/service/search.py

@@ -0,0 +1,81 @@
+from typing import Any
+
+from data_platform_catalogue.search_types import MultiSelectFilter, SortOption
+from django.core.paginator import Paginator
+
+from home.forms.search import SearchForm
+
+from .base import GenericService
+
+
+class SearchService(GenericService):
+    def __init__(self, form: SearchForm, page: str, items_per_page: int = 20):
+        self.form = form
+        self.page = page
+        self.client = self._get_catalogue_client()
+        self.results = self._get_search_results(page, items_per_page)
+        self.paginator = self._get_paginator(items_per_page)
+        self.context = self._get_context()
+
+    def _get_search_results(self, page: str, items_per_page: int):
+        if self.form.is_bound:
+            form_data = self.form.cleaned_data
+        else:
+            form_data = {}
+        query = form_data.get("query", "")
+        sort = form_data.get("sort", "relevance")
+        domains = form_data.get("domains", [])
+        filter_value = [MultiSelectFilter(
+            "domains", domains)] if domains else []
+        page_for_search = str(int(page) - 1)
+        if sort == "ascending":
+            sort_option = SortOption(field="name", ascending=True)
+        elif sort == "descending":
+            sort_option = SortOption(field="name", ascending=False)
+        else:
+            sort_option = None
+
+        results = self.client.search(
+            query=query,
+            page=page_for_search,
+            filters=filter_value,
+            sort=sort_option,
+            count=items_per_page,
+        )
+
+        return results
+
+    def _get_paginator(self, items_per_page: int) -> Paginator:
+        pages_list = list(range(self.results.total_results))
+
+        return Paginator(pages_list, items_per_page)
+
+    def _get_context(self) -> dict[str, Any]:
+
+        if self.form["query"].value:
+            page_title = f'Search for "{self.form["query"].value}" - Data catalogue'
+        else:
+            page_title = "Search - Data catalogue"
+
+        if self.form.is_bound:
+            label_clear_href = {
+                filter.split(":")[-1]: self.form.encode_without_filter(filter)
+                for filter in self.form.cleaned_data.get("domains")
+            }
+        else:
+            label_clear_href = None
+
+        context = {
+            "form": self.form,
+            "results": self.results.page_results,
+            "page_title": page_title,
+            "page_obj": self.paginator.get_page(self.page),
+            "page_range": self.paginator.get_elided_page_range(
+                self.page, on_each_side=2, on_ends=1
+            ),
+            "paginator": self.paginator,
+            "total_results": self.results.total_results,
+            "label_clear_href": label_clear_href,
+        }
+
+        return context

home/templatetags/future.py

@@ -0,0 +1,38 @@
+from django import template
+from django.utils.itercompat import is_iterable
+
+register = template.Library()
+
+
+# Backport from unreleased Django
+# This can be removed after Django 5.1 is released
+@register.simple_tag(takes_context=True)
+def query_string(context, query_dict=None, **kwargs):
+    """
+    Add, remove, and change parameters of a ``QueryDict`` and return the result
+    as a query string. If the ``query_dict`` argument is not provided, default
+    to ``request.GET``.
+    For example::
+        {% query_string foo=3 %}
+    To remove a key::
+        {% query_string foo=None %}
+    To use with pagination::
+        {% query_string page=page_obj.next_page_number %}
+    A custom ``QueryDict`` can also be used::
+        {% query_string my_query_dict foo=3 %}
+    """
+    if query_dict is None:
+        query_dict = context.request.GET
+    query_dict = query_dict.copy()
+    for key, value in kwargs.items():
+        if value is None:
+            if key in query_dict:
+                del query_dict[key]
+        elif is_iterable(value) and not isinstance(value, str):
+            query_dict.setlist(key, value)
+        else:
+            query_dict[key] = value
+    if not query_dict:
+        return ""
+    query_string = query_dict.urlencode()
+    return f"?{query_string}"