Separate the autocompletion logic form the views

shared/autocomplete.py

@@ -0,0 +1,217 @@
+import logging
+from collections import namedtuple
+
+from django.conf import settings
+from django.db.models import Q
+
+if getattr(settings, "LDAP_SERVER_URL", None):
+    import ldap
+else:
+    # shared.tests.testcases.TestCaseMixin.mockLDAP needs
+    # an ldap object to be in the scope
+    ldap = None
+
+
+django_logger = logging.getLogger("django.request")
+
+
+class SearchUnit:
+    """Base class for all the search utilities.
+
+    A search unit should implement a `search` method taking a list of keywords as
+    argument and returning an iterable of search results.
+
+    It might optionally implement the following methods and attributes:
+
+    - result_uuid (method): a callable that takes one result as an input and returns an
+      identifier that is globally unique across search units for this object.
+      This is used to compare results coming from different search units in the
+      `Compose` class. For instance, if the same user can be returned by the LDAP
+      search and a model search instance, using the clipper login as a UUID in both
+      units avoids this user to be returned twice by `Compose`.
+      Returning `None` means that the object should be considered unique.
+    """
+
+    # Mandatory method
+
+    def search(self, _keywords):
+        raise NotImplementedError(
+            "Class implementing the SearchUnit interface should implement the search "
+            "method"
+        )
+
+    # Optional attributes and methods
+
+    def result_uuid(self, result):
+        """A universal unique identifier for the search results."""
+        return None
+
+
+# ---
+# Model-based search
+# ---
+
+
+class ModelSearch(SearchUnit):
+    """Basic search engine for models based on filtering.
+
+    The class should be configured through its `model` class attribute: the `search`
+    method will return a queryset of instances of this model. The `search_fields`
+    attributes indicates which fields to search in.
+
+    Example:
+
+    >>> from django.contrib.auth.models import User
+    >>>
+    >>> class UserSearch(ModelSearch):
+    ...     model = User
+    ...     search_fields = ["username", "first_name", "last_name"]
+    >>>
+    >>> user_search = UserSearch()  # has type ModelSearch[User]
+    >>> user_search.search(["toto", "foo"])  # returns a queryset of Users
+    """
+
+    model = None
+    search_fields = []
+
+    def get_queryset_filter(self, keywords):
+        filter_q = Q()
+
+        if not keywords:
+            return filter_q
+
+        for keyword in keywords:
+            kw_filter = Q()
+            for field in self.search_fields:
+                kw_filter |= Q(**{"{}__icontains".format(field): keyword})
+            filter_q &= kw_filter
+
+        return filter_q
+
+    def search(self, keywords):
+        """Returns the queryset of model instances matching all the keywords.
+
+        The semantic of the search is the following: a model instance appears in the
+        search results iff all of the keywords given as arguments occur in at least one
+        of the search fields.
+        """
+
+        return self.model.objects.filter(self.get_queryset_filter(keywords))
+
+
+# ---
+# LDAP search
+# ---
+
+Clipper = namedtuple("Clipper", ["clipper", "fullname"])
+
+
+class LDAPSearch(SearchUnit):
+    ldap_server_url = getattr(settings, "LDAP_SERVER_URL", None)
+    domain_component = "dc=spi,dc=ens,dc=fr"
+    search_fields = ["cn", "uid"]
+
+    def get_ldap_query(self, keywords):
+        """Return a search query with the following semantics:
+
+        A Clipper appears in the search results iff all of the keywords given as
+        arguments occur in at least one of the search fields.
+        """
+
+        # Dumb but safe
+        keywords = filter(str.isalnum, keywords)
+
+        ldap_filters = []
+
+        for keyword in keywords:
+            ldap_filter = "(|{})".format(
+                "".join(
+                    "({}=*{}*)".format(field, keyword) for field in self.search_fields
+                )
+            )
+            ldap_filters.append(ldap_filter)
+
+        return "(&{})".format("".join(ldap_filters))
+
+    def search(self, keywords):
+        """Return a list of Clipper objects matching all the keywords."""
+
+        query = self.get_ldap_query(keywords)
+
+        if ldap is None or query == "(&)":
+            return []
+
+        try:
+            ldap_obj = ldap.initialize(self.ldap_server_url)
+            res = ldap_obj.search_s(
+                self.domain_component, ldap.SCOPE_SUBTREE, query, self.search_fields
+            )
+            return [
+                Clipper(
+                    clipper=attrs["uid"][0].decode("utf-8"),
+                    fullname=attrs["cn"][0].decode("utf-8"),
+                )
+                for (_, attrs) in res
+            ]
+        except ldap.LDAPError as err:
+            django_logger.error("An LDAP error occurred", exc_info=err)
+            return []
+
+    def result_uuid(self, clipper):
+        return clipper.clipper
+
+
+# ---
+# Composition of autocomplete units
+# ---
+
+
+class Compose:
+    """Search with several units and remove duplicate results.
+
+    The `search_units` class attribute should be a list of pairs of the form `(name,
+    search_unit)`.
+
+    The `search` method produces a dictionary whose keys are the `name`s given in
+    `search_units` and whose values are iterables produced by the different search
+    units.
+
+    Typical Example:
+
+    >>> from django.contrib.auth.models import User
+    >>>
+    >>> class UserSearch(ModelSearch):
+    ...     model = User
+    ...     search_fields = ["username", "first_name", "last_name"]
+    ...
+    ...     def result_uuid(self, user):
+    ...         # Assuming that `.username` stores the clipper login of already
+    ...         # registered users, this avoids showing the same user twice (here and in
+    ...         # then ldap unit).
+    ...         return user.username
+    >>>
+    >>> class UserAndClipperSearch(Compose):
+    ...     search_units = [
+    ...         ("users", UserSearch()),
+    ...         ("clippers", LDAPSearch()),
+    ...     ]
+
+    In this example, clipper accounts that already have an associated user (i.e. with a
+    username equal to the clipper login), will not appear in the results.
+    """
+
+    search_units = []
+
+    def search(self, keywords):
+        seen_uuids = set()
+        results = {}
+        for name, search_unit in self.search_units:
+            uniq_res = []
+            for r in search_unit.search(keywords):
+                uuid = search_unit.result_uuid(r)
+                if uuid is None or uuid not in seen_uuids:
+                    uniq_res.append(r)
+                    if uuid is not None:
+                        seen_uuids.add(uuid)
+            results[name] = uniq_res
+        return results