複雑なORMをSQLで確認する

前回の記事では、Django ORMの基本的な検索を学びました。

Django ORM入門：SQLと比較しながら検索・集計・リレーションを学ぶ

Django ORMと変換後のSQLを並べて比較します。
全件取得から絞り込み、並び替え、JOIN、集計、サブクエリまで段階的に確認していきます。

今回は、複数の条件を組み合わせた検索や、複数階層のリレーションなど、もう少し複雑なクエリを解説します。

掲載するSQLは、MySQLを使用した場合の形を読みやすく整えた例です。DjangoやMySQLのバージョン、QuerySetの組み立て方によって、選択される列、テーブルの別名、条件の並びなどは変わる場合があります。

この記事の環境

$ python --version
Python 3.14.4
$ python -m django --version
6.1
Database
MySQL 8.4.11

使用するモデル

前回使用した従業員と勤怠記録に、会社と部署を加えます。
会社は複数の部署と従業員を持ちます。従業員は必ず会社に所属しますが、入社直後や異動中などの未配属状態を表せるよう、部署は未設定でもよいものとします。さらに、1人の従業員は複数の勤怠記録を持ちます。

erDiagram
    Company ||--o{ Department : "会社に所属"
    Company ||--o{ Employee : "会社に所属"
    Department o|--o{ Employee : "部署に所属"
    Employee ||--o{ Attendance : "勤怠を記録"

    Company {
        bigint id PK "主キー"
        varchar name "会社名"
    }

    Department {
        bigint id PK "主キー"
        bigint company_id FK "会社ID"
        varchar name "部署名"
    }

    Employee {
        bigint id PK "主キー"
        bigint company_id FK "会社ID"
        bigint department_id FK "部署ID・NULL可"
        varchar name "氏名"
        varchar number UK "社員番号"
        int age "年齢"
    }

    Attendance {
        bigint id PK "主キー"
        bigint employee_id FK "従業員ID"
        text note "備考"
        int work_minutes "勤務時間（分）"
        datetime clock_in "出勤日時"
        datetime clock_out "退勤日時・NULL可"
    }
main/models.py
from django.db import models




class Company(models.Model):
    name = models.CharField('会社名', max_length=100)


    class Meta:
        db_table = 'company'
        verbose_name = '会社'




class Department(models.Model):
    company = models.ForeignKey(
        Company,
        on_delete=models.CASCADE,
        related_name='departments',
        verbose_name='会社',
    )
    name = models.CharField('部署名', max_length=100)


    class Meta:
        db_table = 'department'
        verbose_name = '部署'




class Employee(models.Model):
    company = models.ForeignKey(
        Company,
        on_delete=models.PROTECT,
        related_name='employees',
        verbose_name='会社',
    )
    department = models.ForeignKey(
        Department,
        on_delete=models.PROTECT,
        related_name='employees',
        verbose_name='部署',
        null=True,
        blank=True,
    )
    name = models.CharField('氏名', max_length=100)
    number = models.CharField('社員番号', max_length=20, unique=True)
    age = models.IntegerField('年齢')


    class Meta:
        db_table = 'employee'
        verbose_name = '従業員'




class Attendance(models.Model):
    employee = models.ForeignKey(
        Employee,
        on_delete=models.CASCADE,
        related_name='attendances',
        verbose_name='従業員',
    )
    note = models.TextField('備考', blank=True)
    work_minutes = models.IntegerField('勤務時間（分）', default=0)
    clock_in = models.DateTimeField('出勤日時')
    clock_out = models.DateTimeField('退勤日時', null=True, blank=True)


    class Meta:
        db_table = 'attendance'
        verbose_name = '勤怠記録'

related_nameは、関連先から逆方向へ参照するときの名前です。例えば、従業員から勤怠記録をたどる場合はemployee.attendances.all()、QuerySetの検索条件ではattendances__clock_outのように指定できます。

Employee.companyとEmployee.department.companyは、外部キーだけでは同じ会社になることを保証できません。従業員へ部署を設定するときは、その部署がEmployee.companyと同じ会社に属しているかをフォーム、モデルの検証、または更新処理で確認します。

1. Qオブジェクトをネストして複雑な条件を作る

サンプル株式会社に所属する20歳以上60歳未満の従業員のうち、開発部に所属している、または部署が未設定である従業員を検索します。

Django ORM
from django.db.models import Q


employees = Employee.objects.filter(
    Q(company__name='サンプル株式会社')
    & Q(age__gte=20)
    & Q(age__lt=60)
    & (
        Q(department__name='開発部')
        | Q(department__isnull=True)
    )
)
SQL
SELECT `employee`.*
FROM `employee`
INNER JOIN `company`
    ON `employee`.`company_id` = `company`.`id`
LEFT OUTER JOIN `department`
    ON `employee`.`department_id` = `department`.`id`
WHERE (
    `company`.`name` = 'サンプル株式会社'
    AND `employee`.`age` >= 20
    AND `employee`.`age` < 60
    AND (
        `department`.`name` = '開発部'
        OR `employee`.`department_id` IS NULL
    )
);

Qオブジェクトは、&でAND、|でOR、~でNOTを表します。ネストした括弧はSQLの条件にも反映されます。

条件が長くなる場合は、目的ごとに変数へ分けると読みやすくなります。
下記のようにしても検索条件は変わりません。

working_age = Q(age__gte=20) & Q(age__lt=60)


target_department = (
    Q(department__name='開発部')
    | Q(department__isnull=True)
)


employees = Employee.objects.filter(
    Q(company__name='サンプル株式会社')
    & working_age
    & target_department
)

2. 複数階層のリレーションをたどる

サンプル株式会社の開発部に所属する従業員のうち、退勤していない勤怠記録を取得します。

Django ORM
attendances = Attendance.objects.filter(
    employee__company__name='サンプル株式会社',
    employee__department__name='開発部',
    clock_out__isnull=True,
)
SQL
SELECT `attendance`.*
FROM `attendance`
INNER JOIN `employee`
    ON `attendance`.`employee_id` = `employee`.`id`
INNER JOIN `department`
    ON `employee`.`department_id` = `department`.`id`
INNER JOIN `company`
    ON `employee`.`company_id` = `company`.`id`
WHERE (
    `company`.`name` = 'サンプル株式会社'
    AND `department`.`name` = '開発部'
    AND `attendance`.`clock_out` IS NULL
);

employee__company__nameは勤怠記録から従業員、会社へ、employee__department__nameは勤怠記録から従業員、部署へ関連をたどっています。

Employee.departmentにはnull=Trueを指定しているため、部署が未設定の従業員が存在する可能性があります。ただし、この検索ではemployee__department__name='開発部'を条件にしているため、INNER JOINとなり、関連する部署が存在しない従業員は結果に含まれません。

上記の例ではEmployeeやDepartmentはSELECTされないので、ループ内で会社名や部署名へアクセスすると追加クエリが発生します。
取得後に関連オブジェクトも参照する場合は、select_related()を指定します。

attendances = Attendance.objects.select_related(
    'employee__company',
    'employee__department',
).filter(
    employee__company__name='サンプル株式会社',
    employee__department__name='開発部',
    clock_out__isnull=True,
)


for attendance in attendances:
    print(
        attendance.employee.name,
        attendance.employee.department.name,
        attendance.employee.company.name,
    )

3. 従業員から勤怠記録をたどって複数条件を指定する

Attendance.employeeという外部キーを、従業員側からattendancesを使ってたどることを「逆方向の参照」と呼びます。この関連に複数の条件を指定すると、条件の書き方によって意味が変わります。
次の例は、「勤務時間が480分以上で、なおかつ未退勤である同じ勤怠記録」を持つ従業員を検索します。

同じ勤怠記録が両方の条件を満たす

Django ORM
employees = Employee.objects.filter(
    attendances__work_minutes__gte=480,
    attendances__clock_out__isnull=True,
)
SQL
SELECT `employee`.*
FROM `employee`
INNER JOIN `attendance`
    ON `employee`.`id` = `attendance`.`employee_id`
WHERE (
    `attendance`.`work_minutes` >= 480
    AND `attendance`.`clock_out` IS NULL
);

1回のfilter()に条件をまとめているため、同じattendanceテーブルの行に対して両方の条件が適用されます。

一方、次のようにfilter()を分けると意味が変わります。

別々の勤怠記録が条件を満たしても一致する

Django ORM
employees = Employee.objects.filter(
    attendances__work_minutes__gte=480,
).filter(
    attendances__clock_out__isnull=True,
)
SQL
SELECT `employee`.*
FROM `employee`
INNER JOIN `attendance`
    ON `employee`.`id` = `attendance`.`employee_id`
INNER JOIN `attendance` `T3`
    ON `employee`.`id` = `T3`.`employee_id`
WHERE (
    `attendance`.`work_minutes` >= 480
    AND `T3`.`clock_out` IS NULL
);

このSQLではattendanceが2回JOINされています。そのため、「480分以上働いた勤怠記録」と「未退勤の勤怠記録」が別の日のレコードでも従業員が一致します。

また、1人の従業員に条件を満たす勤怠記録が複数あると、JOIN後の結果に同じ従業員が複数回現れることがあります。従業員を重複させたくない場合はdistinct()を使用します。

employees = Employee.objects.filter(
    attendances__clock_out__isnull=True,
).distinct()
SELECT DISTINCT `employee`.*
FROM `employee`
INNER JOIN `attendance`
    ON `employee`.`id` = `attendance`.`employee_id`
WHERE `attendance`.`clock_out` IS NULL;

4. 未退勤の件数と勤務時間の合計で従業員を絞り込む

各従業員に、未退勤の勤怠件数と勤務時間の合計を追加します。そのうえで、未退勤の勤怠が1件以上あり、合計勤務時間が2400分以上の従業員だけを取得します。

Django ORM
from django.db.models import Count, Q, Sum


employees = Employee.objects.annotate(
    open_count=Count(
        'attendances',
        filter=Q(attendances__clock_out__isnull=True),
    ),
    total_work_minutes=Sum(
        'attendances__work_minutes',
        default=0,
    ),
).filter(
    open_count__gte=1,
    total_work_minutes__gte=2400,
)
SQL
SELECT
    `employee`.*,
    COUNT(
        CASE
            WHEN `attendance`.`clock_out` IS NULL
            THEN `attendance`.`id`
            ELSE NULL
        END
    ) AS `open_count`,
    COALESCE(
        SUM(`attendance`.`work_minutes`),
        0
    ) AS `total_work_minutes`
FROM `employee`
LEFT OUTER JOIN `attendance`
    ON `employee`.`id` = `attendance`.`employee_id`
GROUP BY `employee`.`id`
HAVING (
    COUNT(
        CASE
            WHEN `attendance`.`clock_out` IS NULL
            THEN `attendance`.`id`
            ELSE NULL
        END
    ) >= 1
    AND COALESCE(
        SUM(`attendance`.`work_minutes`),
        0
    ) >= 2400
)
ORDER BY NULL;

集計関数のfilterは、その集計に含める行だけを指定します。ここでは、Count()だけが未退勤の勤怠記録を対象にし、Sum()はすべての勤怠記録を合計しています。

MySQLは集計関数のFILTER (WHERE ...)構文をサポートしていません。そのためDjangoは条件付きのCount()をCASE WHENを使った形へ変換します。

また、annotate()で追加した集計結果に対する絞り込みは、SQLではWHEREではなくHAVINGになります。WHEREは集計前の行を絞り込み、HAVINGはグループ化と集計が終わった結果を絞り込みます。

annotate()とfilter()の順番

次の2つは同じ意味ではありません。

先に対象期間を絞ってから集計する
Django ORM
Employee.objects.filter(
    attendances__clock_in__year=2026,
).annotate(
    total=Sum('attendances__work_minutes'),
)
全期間を集計してから2026年の勤怠がある人を探す
Django ORM
Employee.objects.annotate(
    total=Sum('attendances__work_minutes'),
).filter(
    attendances__clock_in__year=2026,
)

前者のtotalは2026年の勤務時間だけを合計します。後者のtotalは全期間の勤務時間を合計し、別のJOINを使って2026年の勤怠があるかを調べます。

QuerySetのメソッドは、記述した順番でSQLの構造へ影響します。集計結果がおかしいときは、filter()が集計の前と後のどちらにあるか、生成されたJOINが増えていないかを確認してください。

5. Subqueryで最新の勤怠時刻を追加する

Exists()は関連レコードが存在するかどうかを真偽値で調べるのに向いていますが、関連レコードから実際の値を1つ取得したい場合はSubquery()を使います。

各従業員に、最新の出勤日時をlatest_clock_inとして追加します。

Django ORM
from django.db.models import OuterRef, Subquery


latest_attendance = Attendance.objects.filter(
    employee=OuterRef('pk'),
).order_by('-clock_in')


employees = Employee.objects.annotate(
    latest_clock_in=Subquery(
        latest_attendance.values('clock_in')[:1]
    )
)
SQL
SELECT
    `employee`.*,
    (
        SELECT `U0`.`clock_in`
        FROM `attendance` `U0`
        WHERE `U0`.`employee_id` = `employee`.`id`
        ORDER BY 1 DESC
        LIMIT 1
    ) AS `latest_clock_in`
FROM `employee`;

OuterRef('pk')は、外側のSELECTで処理している従業員の主キーを参照します。従業員ごとに対応する勤怠記録を調べ、出勤日時の新しい順に並べた先頭の値を返します。

Subquery()へ渡すQuerySetでは、次の2点が重要です。

get()はその場でデータベースへ問い合わせ、1件を取得しようとします。しかし、この時点ではOuterRef('pk')が参照する従業員がまだ決まっていないため、取得できません。[:1]なら問い合わせをすぐには実行せず、サブクエリにLIMIT 1を付けられます。

勤怠記録がない従業員のlatest_clock_inはNoneになります。

6. F、Case、Whenでデータベース上の値を計算する

1日の勤務時間が480分を超えた分を、残業時間として追加します。

Django ORM
from django.db.models import Case, F, IntegerField, Value, When


attendances = Attendance.objects.annotate(
    overtime_minutes=Case(
        When(
            work_minutes__gt=480,
            then=F('work_minutes') - 480,
        ),
        default=Value(0),
        output_field=IntegerField(),
    )
)
SQL
SELECT
    `attendance`.*,
    CASE
        WHEN `attendance`.`work_minutes` > 480
        THEN (`attendance`.`work_minutes` - 480)
        ELSE 0
    END AS `overtime_minutes`
FROM `attendance`;

F('work_minutes')は、現在の勤怠記録のwork_minutes列を計算式の中で参照します。Case()とWhen()はSQLのCASE WHENに対応します。

F()はフィールド同士の比較にも使えます。例えば、退勤日時が出勤日時より前になっている異常なデータは次のように検索できます。

invalid_attendances = Attendance.objects.filter(
    clock_out__lt=F('clock_in')
)
SELECT `attendance`.*
FROM `attendance`
WHERE `attendance`.`clock_out` < `attendance`.`clock_in`;

値をPythonへ読み込んで1件ずつ比較する必要がなく、条件判定をデータベース側で行えます。

7. select_relatedとPrefetchを組み合わせる

会社、部署、従業員、未退勤の勤怠記録を一覧表示するとします。

DepartmentやCompanyはForeignKeyの関連先なのでselect_related()で取得できます。しかし、1人の従業員が複数持つ勤怠記録は、同じ方法では事前取得できません。

複数の関連オブジェクトを取得するときはprefetch_related()を使います。さらにPrefetchを使うと、関連先のQuerySetを絞り込めます。

Django ORM
from django.db.models import Prefetch


employees = Employee.objects.select_related(
    'company',
    'department',
).prefetch_related(
    Prefetch(
        'attendances',
        queryset=Attendance.objects.filter(
            clock_out__isnull=True,
        ).order_by('-clock_in'),
        to_attr='open_attendances',
    )
)

このQuerySetは、次の2つのSELECT文が実行されます。

1. 従業員・部署・会社を取得

SQL
SELECT
    `employee`.*,
    `department`.*,
    `company`.*
FROM `employee`
LEFT OUTER JOIN `department`
    ON `employee`.`department_id` = `department`.`id`
INNER JOIN `company`
    ON `employee`.`company_id` = `company`.`id`;

2. 対象従業員の未退勤レコードをまとめて取得

SQL
SELECT `attendance`.*
FROM `attendance`
WHERE (
    `attendance`.`clock_out` IS NULL
    AND `attendance`.`employee_id` IN (1, 2, 3, ...)
)
ORDER BY `attendance`.`clock_in` DESC;

select_related()はSQLのJOINで単一の関連オブジェクトを同時に取得します。prefetch_related()は関連先を別のSQLでまとめて取得し、DjangoがPython側で対応付けます。

to_attr='open_attendances'を指定すると、絞り込んで取得した勤怠記録を、各Employeeオブジェクトのopen_attendances属性からリストとして参照できます。

for employee in employees:
    print(employee.company.name)
    if employee.department is not None:
        print(employee.department.name)
    print(employee.name)


    for attendance in employee.open_attendances:
        print(attendance.clock_in)

部署が未設定の場合はemployee.departmentがNoneになるため、部署名へアクセスする前に確認します。ループ内で会社、部署、未退勤の勤怠記録へアクセスしても、従業員ごとの追加問い合わせは発生しません。部署は任意の関連なので、生成されるSQLではdepartmentがLEFT OUTER JOINで結合されます。

モデル定義のAttendance.employeeにはrelated_name='attendances'を指定しています。そのため、Employeeオブジェクトからemployee.attendancesで関連する勤怠記録を扱えます。

employee.attendancesは、関連する勤怠記録を検索するためにDjangoが用意する「関連マネージャー」です。employee.attendances.all()は、その従業員のすべての勤怠記録を取得するQuerySetを返します。

今回to_attrで事前取得した未退勤の記録は、別のopen_attendances属性から参照します。事前取得した結果を使うときはemployee.open_attendancesへアクセスします。

複雑なQuerySetを確認するときのポイント

複雑なQuerySetを作ったときは、ORMの行数だけで効率を判断せず、次の点を確認します。

確認すること 主な確認方法
どのSQLへ変換されたか print(queryset.query)
JOINが意図せず増えていないか 生成されたSQLのJOINとテーブル別名を確認
同じモデルが重複していないか 件数と主キーを確認し、必要ならdistinct()を検討
集計前と集計後のどちらで絞っているか WHEREとHAVINGを確認
N+1が発生していないか CaptureQueriesContextやassertNumQueries()でクエリ数を確認
インデックスが使われているか queryset.explain()で実行計画を確認

特に、勤怠記録を逆方向にたどる複数条件、annotate()とfilter()の順番、select_related()とprefetch_related()の使い分けは、結果やクエリ数が変わりやすい部分です。

ORMで複雑な問い合わせを表現できても、それだけで正しい結果や十分な性能が保証されるわけではありません。小さなテストデータだけでなく、条件を満たす関連レコードが複数ある場合や、関連レコードが1件もない場合も確認しておくと、JOINや集計による見落としを減らせます。