Кастомные 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".