Кастомные lookup-операторы в Django ORM

Привет, Хабр!

Сегодня рассмотрим тему кастомных lookup‑операторов в Django ORM. Они позволяют расширить стандартный синтаксис Django, интегрируя свои SQL‑функции и алгоритмы, при этом сохраняя привычный вид фильтрации.

Обзор синтаксиса кастомных lookup-операторов

Каждый кастомный lookup в Django — это класс, наследник django.db.models.lookups.Lookup. Основная его задача — реализовать метод as_sql(), который генерирует SQL‑код для запроса. Пример схемы:

from django.db.models import Lookup, Field  class MyCustomLookup(Lookup):     # Имя lookup-а, используемое в фильтрах (например, __mylookup)     lookup_name = 'mylookup'      def as_sql(self, compiler, connection):         # process_lhs() преобразует левую часть выражения (поле модели) в SQL         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         # process_rhs() делает то же самое для правой части (значение фильтра)         rhs_sql, rhs_params = self.process_rhs(compiler, connection)         # Собираем итоговое SQL-условие. Здесь можно применять SQL-функции, операторы и т.д.         sql = "%s = %s" % (lhs_sql, rhs_sql)         params = lhs_params + rhs_params         return sql, params  # Регистрация lookup-а для нужного типа поля или для всех полей Field.register_lookup(MyCustomLookup)

Основные моменты. lookup_name: это имя, под которым вы будете вызывать оператор, например, field__mylookup=value. process_lhs() и process_rhs(): гарантируют, что входные данные корректно экранируются и адаптируются под конкретную СУБД. as_sql(): здесь формируется SQL‑условие, используя компоненты запроса. Можно вызывать встроенные SQL‑функции, комбинировать выражения и делать подзапросы.

Примеры применения

Lookup для звукового поиска

Допустим, есть модель для хранения имен:

# models.py from django.db import models  class Person(models.Model):     name = models.CharField(max_length=255)      def __str__(self):         return self.name

Создадим lookup, который использует функцию SOUNDEX для поиска похожих по звучанию имен:

# lookups.py from django.db.models import Lookup, Field  class SoundexLookup(Lookup):     lookup_name = 'soundex'      def as_sql(self, compiler, connection):         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         rhs_sql, rhs_params = self.process_rhs(compiler, connection)         # Генерируем SQL-условие: сравниваем SOUNDEX от поля и от переданного значения         sql = "SOUNDEX(%s) = SOUNDEX(%s)" % (lhs_sql, rhs_sql)         params = lhs_params + rhs_params         return sql, params  Field.register_lookup(SoundexLookup)

Используем его в запросе:

from myapp.models import Person  # Если в базе "John", "Jon" и "Juan" — оператор найдёт нужные варианты matching_people = Person.objects.filter(name__soundex="John") for person in matching_people:     print(f"Найдено: {person.name}")

Если вы используете PostgreSQL, проверьте наличие расширения fuzzystrmatch.

Полнотекстовый поиск с to_tsvector/to_tsquery

Переходим к более сложной задаче — поиску по тексту. Представим модель статьи:

# models.py from django.db import models  class Article(models.Model):     title = models.CharField(max_length=255)     content = models.TextField()      def __str__(self):         return self.title

Создадим lookup для полнотекстового поиска:

# lookups.py from django.db.models import Lookup, Field  class FullTextSearchLookup(Lookup):     lookup_name = 'fts'      def as_sql(self, compiler, connection):         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         rhs_sql, rhs_params = self.process_rhs(compiler, connection)         # 'russian' — конфигурация языка         sql = "to_tsvector('russian', %s) @@ to_tsquery('russian', %s)" % (lhs_sql, rhs_sql)         params = lhs_params + rhs_params         return sql, params  Field.register_lookup(FullTextSearchLookup)

Пример запроса:

from myapp.models import Article  # Найдёт статьи, где в тексте встречается слово "Django" articles = Article.objects.filter(content__fts="Django") for article in articles:     print(f"Статья: {article.title}")

Чтобы ускорить поиск, можно создать индекс:

CREATE INDEX article_content_idx ON myapp_article USING gin(to_tsvector('russian', content));

Геопространственный lookup с PostGIS

Для проектов, где нужны геоданные, GeoDjango и PostGIS есть множество возможностей. Пример модели:

# models.py from django.contrib.gis.db import models as geomodels  class Location(geomodels.Model):     name = geomodels.CharField(max_length=255)     coordinates = geomodels.PointField(geography=True)      def __str__(self):         return self.name

Создадим lookup для поиска объектов в пределах заданного радиуса с помощью ST_Distance:

# lookups.py from django.db.models import Lookup, Field  class STDistanceLessThan(Lookup):     lookup_name = 'st_dlt'      def as_sql(self, compiler, connection):         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         # Ожидаем, что rhs — это кортеж (target_point, threshold)         if not isinstance(self.rhs, (tuple, list)) or len(self.rhs) != 2:             raise ValueError("Для st_dlt требуется кортеж (target_point, threshold)")         target_point, threshold = self.rhs         sql = "ST_Distance(%s, %%s) < %%s" % lhs_sql         params = lhs_params + [target_point, threshold]         return sql, params  Field.register_lookup(STDistanceLessThan)

Пример запроса:

from django.contrib.gis.geos import Point from myapp.models import Location  # Координаты центра Москвы center = Point(37.6173, 55.7558) radius = 1000  # в метрах  nearby_locations = Location.objects.filter(coordinates__st_dlt=(center, radius)) for loc in nearby_locations:     print(f"{loc.name} находится в пределах {radius} м от центра Москвы")

Lookup для поиска с алгоритмом Левенштейна и Regex

Чтобы обрабатывать опечатки и находить похожие строки, можно использовать алгоритм Левенштейна. Рассмотрим модель продукта:

# models.py from django.db import models  class Product(models.Model):     name = models.CharField(max_length=255)      def __str__(self):         return self.name

Lookup для Левенштейна:

# lookups.py from django.db.models import Lookup, Field  class LevenshteinLookup(Lookup):     lookup_name = 'lev'      def as_sql(self, compiler, connection):         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         rhs_sql, rhs_params = self.process_rhs(compiler, connection)         if not isinstance(self.rhs, (tuple, list)) or len(self.rhs) != 2:             raise ValueError("Для lev требуется кортеж (искомая строка, порог)")         search_str, threshold = self.rhs         sql = "levenshtein(%s, %%s) <= %%s" % lhs_sql         params = lhs_params + [search_str, threshold]         return sql, params  Field.register_lookup(LevenshteinLookup)

Пример использования:

from myapp.models import Product  # Находим товары, где название отличается от "Smartphone" не более чем на 2 символа similar_products = Product.objects.filter(name__lev=("Smartphone", 2)) for prod in similar_products:     print(f"Похожий продукт: {prod.name}")

А чтобы добавить универсальность, можно написать lookup для поиска по регулярке. Допустим, есть модель комментариев:

# models.py from django.db import models  class Comment(models.Model):     author = models.CharField(max_length=100)     content = models.TextField()      def __str__(self):         return f"{self.author}: {self.content[:20]}..."

Lookup для Regex:

# lookups.py from django.db.models import Lookup, Field  class RegexLookup(Lookup):     lookup_name = 'regex'      def as_sql(self, compiler, connection):         lhs_sql, lhs_params = self.process_lhs(compiler, connection)         rhs_sql, rhs_params = self.process_rhs(compiler, connection)         sql = "%s ~ %s" % (lhs_sql, rhs_sql)         params = lhs_params + rhs_params         return sql, params  Field.register_lookup(RegexLookup)

Пример запроса:

from myapp.models import Comment  # Фильтруем комментарии, где встречается слово, начинающееся на "django" matching_comments = Comment.objects.filter(content__regex=r'\bdjango\w*') for comment in matching_comments:     print(f"Комментарий: {comment.content}")

Тестирование

Ни один оператор не обходится без тестов! Пример набора тестов для lookup‑операторов:

# tests.py from django.test import TestCase from myapp.models import Person, Product, Article, Comment, Location from django.contrib.gis.geos import Point  class LookupTests(TestCase):     def setUp(self):         Person.objects.create(name="John")         Person.objects.create(name="Jon")         Person.objects.create(name="Juan")                  Product.objects.create(name="Smartphone")         Product.objects.create(name="Smartfone")         Product.objects.create(name="Smatphone")                  Article.objects.create(title="Django Tips", content="Полнотекстовый поиск в Django с использованием PostgreSQL")         Article.objects.create(title="ORM магия", content="Расширяем возможности Django ORM через кастомные lookup-операторы.")                  Comment.objects.create(author="Alice", content="Django — это круто!")         Comment.objects.create(author="Bob", content="Я люблю django-разработку.")                  Location.objects.create(name="Центр", coordinates=Point(37.6173, 55.7558))         Location.objects.create(name="Окрестности", coordinates=Point(37.6300, 55.7600))      def test_soundex_lookup(self):         qs = Person.objects.filter(name__soundex="John")         self.assertEqual(qs.count(), 2)      def test_levenshtein_lookup(self):         qs = Product.objects.filter(name__lev=("Smartphone", 2))         self.assertGreaterEqual(qs.count(), 1)      def test_fulltext_search_lookup(self):         qs = Article.objects.filter(content__fts="Django")         self.assertGreaterEqual(qs.count(), 1)      def test_regex_lookup(self):         qs = Comment.objects.filter(content__regex=r'\bdjango\w*')         self.assertGreaterEqual(qs.count(), 1)      def test_st_distance_lookup(self):         center = Point(37.6173, 55.7558)         qs = Location.objects.filter(coordinates__st_dlt=(center, 1000))         self.assertGreaterEqual(qs.count(), 1)

---

Если у вас есть интересные кейсы применения и хочется поделиться своим опытом, пишите в комментариях.

20 февраля в Otus пройдёт открытый урок на тему «Децентрализованная революция в управлении данными: Data Mesh и его четыре принципа».

Если тема для вас актуальна, записывайтесь на странице курса «Data Engineer».