Фильтрация Django Queryset на основе каждой записи

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

class Travel(models.Model):
  car = models.CharField()
  road = models.CharField()
  start = models.DateTimeField()
  end = models.DateTimeField()

Я хочу определить множество автомобилей X, которые находились на той же дороге, что и целевой автомобиль x в течение не менее m минут.

Как получить требуемый набор автомобилей X?

Моя попытка:

Допустим, я использую фильтрацию для получения набора путешествий T, в которых побывал x.

T <-- Travel.objects.filter(car=x)

Затем я перебором получаю: 

for t in T:
  possible_travels <-- filter Travel.objects with car=/=x, road=t.road, start < t.end, end > t.start 
  
  confirmed_travels <-- further filter possible_travels with the overlapping region being at least m minutes long

  confirmed_cars <-- confirmed_travels.values('cars').distinct()

Однако проблема заключается в том, что это может повлечь за собой множество обращений к БД при запросе в цикле. Кроме того, confirmed_cars дает объект QuerySet. Похоже, что мне нужно как-то объединить эти объекты QuerySet вместе. Я видел другие сообщения, в которых делались такие вещи, как преобразование в список, затем добавление и, наконец, преобразование обратно в QuerySet, но некоторые люди говорят, что это не очень хороший способ, должен ли я делать что-то подобное?

Есть ли какие-то лучшие подходы к этому? Действительно ли цикл for необходим и можно ли обойтись без него?

Вот мое решение с кверисетами и без каких-либо циклов. Для того чтобы найти пересечение между start и end временами, я разбил кверисет на 4 категории:

  • start1 -> start2 -> end1 -> end2 (q_slides_before) (пересечение = end1 - start2)
  • start1 -> start2 -> end2 -> end1 (q_contains) (intersection = end2 - start2)
  • start2 -> start1 -> end1 -> end2 (q_is_contained) (intersection = end1 - start1)
  • start2 -> start1 -> end2 -> end1 (q_slides_after) (intersection = end2 - start1)
from django.db.models import OuterRef, ExpressionWrapper,\
                             F, Q, functions, DurationField, FloatField

# Use this function instead of "annotate_delta" if your db supports DurationField (postgres supports this but sqlite does not)
def annotate_delta_with_duration_support(qs, start, end):
    duration_exp = ExpressionWrapper(end - start, output_field=DurationField())
    return qs.annotate(delta=functions.ExtractMinute(duration_exp))


def annotate_delta(qs, start, end):
    duration_exp = ExpressionWrapper((end - start) / (60 * 10**6),
                                     output_field=FloatField())
    return qs.annotate(delta=duration_exp)


x = 'mycar'
m = 20

q_is_contained = Q(start__gte=OuterRef('start')) & Q(end__lte=OuterRef('end'))
qs = annotate_delta(Travel.objects, F('start'), F('end'))
qs = qs.filter(q_is_contained, delta__gte=m, car=x, road=OuterRef('road'))
res1 = Travel.objects.exclude(car=x).annotate(ex=Exists(qs)).filter(ex=True)


q_contains = Q(start__lte=OuterRef('start')) & Q(end__gte=OuterRef('end'))
qs = annotate_delta(Travel.objects, OuterRef('start'), OuterRef('end'))
qs = qs.filter(q_contains, delta__gte=m, car=x, road=OuterRef('road'))
res2 = Travel.objects.exclude(car=x).annotate(ex=Exists(qs)).filter(ex=True)


q_slides_before = Q(start__lte=OuterRef('start')) & \
                  Q(end__lte=OuterRef('end')) & \
                  Q(end__gte=OuterRef('start'))
qs = annotate_delta(Travel.objects, OuterRef('start'), F('end'))
qs = qs.filter(q_slides_before, delta__gte=m, car=x, road=OuterRef('road'))
res3 = Travel.objects.exclude(car=x).annotate(ex=Exists(qs)).filter(ex=True)


q_slides_after = Q(start__gte=OuterRef('start')) & \
                 Q(end__gte=OuterRef('end')) & \
                 Q(start__lte=OuterRef('end'))
qs = annotate_delta(Travel.objects, F('start'), OuterRef('end'))
qs = qs.filter(q_slides_after, delta__gte=m, car=x, road=OuterRef('road'))
res4 = Travel.objects.exclude(car=x).annotate(ex=Exists(qs)).filter(ex=True)

res = res1 | res2 | res3 | res4

Если вы не пишете многоразовую библиотеку и точно застряли с каким-то конкретным бэкендом базы данных, вы можете использовать и необработанный SQL. Это не лучший выбор, но если производительность этого запроса является узким местом - одиночный запрос может быть быстрее. Что-то вроде этого (не проверено, но должно работать в postgres, нужно немного изменить для работы с другими бэкендами):

from django.db import connection
with connection.cursor() as cur:
    cur.execute('''
        SELECT others.car
        FROM {table} base
            INNER JOIN {table} as others
                on base.car <> others.car
                AND ((LEAST(base.end, others.end) - GREATEST(base.start, others.start)) > interval '%s minutes')
        WHERE base.car = %s
    '''.format(table=Travel._meta.db_table), (5, 'car1'))
    cur.fetchall()  # should return list of 1-tuples with proper cars
Вернуться на верх