from collections import namedtuple from dal import autocomplete from django.conf import settings from django.db.models import Q if getattr(settings, "LDAP_SERVER_URL", None): from ldap3 import Connection else: # shared.tests.testcases.TestCaseMixin.mockLDAP needs # Connection to be defined Connection = None 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. """ def search(self, _keywords): raise NotImplementedError( "Class implementing the SeachUnit interface should implement the search " "method" ) # --- # 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)) class Select2QuerySetView(ModelSearch, autocomplete.Select2QuerySetView): """Compatibility layer between ModelSearch and Select2QuerySetView.""" def get_queryset(self): keywords = self.q.split() return super().search(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): # 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. The semantic of the search is the following: a Clipper appears in the search results iff all of the keywords given as arguments occur in at least one of the search fields. """ query = self.get_ldap_query(keywords) if Connection is None or query == "(&)": return [] with Connection(self.ldap_server_url) as conn: conn.search(self.domain_component, query, attributes=self.search_fields) return [Clipper(entry.uid.value, entry.cn.value) for entry in conn.entries] # --- # Composition of autocomplete units # --- class Compose: """Search with several units and remove duplicate results. The ``search_units`` class attribute should be a list of tuples of the form ``(name, uniq_key, search_unit)``. The ``search`` method produces a dictionnary whose keys are the ``name``s given in ``search_units`` and whose values are iterables produced by the different search units. The ``uniq_key``s are used to remove duplicates: for instance, say that search unit 1 has ``uniq_key = "username"`` and search unit 2 has ``uniq_key = "clipper"``, then search results from unit 2 whose ``.clipper`` attribute is equal to the ``.username`` attribute of some result from unit 1 are omitted. Typical Example: >>> from django.contrib.auth.models import User >>> >>> class UserSearch(ModelSearch): ... model = User ... search_fields = ["username", "first_name", "last_name"] >>> >>> class UserAndClipperSearch(Compose): ... search_units = [ ... ("users", "username", UserSearch), ... ("clippers", "clipper", 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): uniq_results = set() results = {} for name, uniq_key, search_unit in self.search_units: res = search_unit().search(keywords) res = [r for r in res if getattr(r, uniq_key) not in uniq_results] uniq_results |= set((getattr(r, uniq_key) for r in res)) results[name] = res return results