Django 5系で silently fail - “空のクエリ問題”に注意
bubusuke です。
2025年4月にDjango 5系のLTS(Long Term Support)版であるDjango 5.2が出ましたね。
WiseVineのプロダクトではDjango4.2を使っていますが、Django4.2は2026年4月末でセキュリティサポートが終了します。
そろそろ切り替えようかと思い準備を進めていた矢先、意図せぬ挙動に遭遇しました。
この挙動に気づかないと、本番でデータ抽出が silently fail する危険があります。
公式に連絡しましたが対応しないという回答だったので、注意喚起として発信します。
※今回記載する内容以外は大きな変更なく対応できそうでした。
意図せぬ挙動
後方互換性のない変更として記載されている内容です。
Filtering querysets against overflowing integer values now always returns an empty queryset. As a consequence, you may need to use ExpressionWrapper() to explicitly wrap arithmetic against integer fields in such cases.
訳:整数値のオーバーフローに対するクエリセットのフィルタリングは、常に空のクエリセットを返すようになりました。このため、このようなケースでは整数フィールドに対する算術演算を明示的にラップするためにExpressionWrapper()を使用する必要が生じる可能性があります。
...記載されているとおり、エラーになるわけでもなく、「ただただ結果が空になる」ということが起きます。いっそのこと、エラーにしてくれたらいいのに...
具体的に再現してみましょう。
ContentTypeというDjangoModelのモデル数分レコードがあるものに対して必ず正となる条件(ただしオーバーフローが発生するもの)を指定してクエリ実行してみます。
Django5系では以下のコードでcountが0件になります。
Django4系ではモデルの数のcountを取得できます。
def test_integer_overflow():
num = 10000000000 # integer だとオーバーフローする内容
qs = ContentType.objects.alias( # 確認のためにContentType.objects.aliasを使ってますが、業務テーブルの項目として捉えてください。.
after=Value(100000000000, output_field=BigIntegerField()), # numよりも1桁大きくしてます
before=Value(9000, output_field=BigIntegerField()),
diff=F("after") - F("before"),
).filter(diff__gt=num) # diff>num が成り立つのでTrueのはず。
# Django 5.x: オーバーフロー発生 → 空のクエリセット(0件)
# Django 4.x: すべてのContentTypeが取得される(count > 0)
assert qs.count() > 0
以下のように ExpressionWrapper などを使いつつ型を指定するとDjango5系でも想定通りのcount結果を取得できます。
def test_not_integer_overflow():
num = 10000000000
qs = ContentType.objects.alias(
after=Value(100000000000, output_field=BigIntegerField()),
before=Value(9000, output_field=BigIntegerField()),
# ↓ を変えました
diff=ExpressionWrapper(F("after") - F("before"), output_field=BigIntegerField()),
).filter(diff__gt=num)
assert qs.count() > 0 # Success
対処の難しさ
再現させた時のように、「ExpressionWrapperで明示的に型指定」すれば解消します。
しかし、「漏れなくExpressionWrapperを入れられるか」という点が問題です。
この挙動の問題点は silently fail であることです。気付けない...
漏れなく検知するいい方法は、残念ながらまだ見つけられていません。 2025/10/23 15:30 UPDATE! 解決策見つけました!(後述)
EmptyResultSetはDjango内部でexceptされ、アプリケーション側に到達するまでに揉み消されてしまうため、検知自体ができません。
別のExceptionにしてアプリケーション側までRaiseされるような提案もしましたが、採用されませんでした。
「生成AIのレビュー観点にしてもらう」、「Pylintの静的チェック」などで検知できるか試してみてますが、いまのところうまくいっていないです(Pylintは偽陽性も多く引っかかるようなものに)。
「気にせずもう切り替えちゃう」、「forkして前述の『別のExceptionにしてアプリケーション側までRaiseする』仕組みを入れる」など、いろんな選択肢を出して検討しているところです。
対処方法(弊社の場合)
前述のように対処が難しいのですが、弊社では ExpressionWrapperなしで四則演算することを禁止する という対応をとることにしました。
これによって、alias, annotate, filter 利用時にExpressionWrapperなしで四則演算されているものがあれば全て実行時にエラーになります。Sum(hoge, filter=...)などの場合も hogeの中のCaseや、filterの中でCaseやSumなどもあっても再帰的にチェックが走ります。
手順
-
アプリケーション全体で一律利用するBaseQuerySetを作成。
-
アプリケーションの全てのDjangoModelのQuerySetがBaseQuerySetを使うように設定。
※Modelが漏れなくBaseQuerySetを利用することを担保する仕組みはまた別の記事で。 -
BaseQuerySetの以下のコードを記載する。
コード
from __future__ import annotations
from typing import Self
from django.db import models
from django.db.models import Case, ExpressionWrapper, When
from django.db.models.aggregates import Aggregate
from django.db.models.expressions import CombinedExpression
class BaseQuerySet(models.QuerySet):
def annotate(self, *args, **kwargs) -> Self:
for name, expression in kwargs.items():
self._check_expression_recursive(expression, context=f"annotate '{name}'")
return super().annotate(*args, **kwargs)
def alias(self, *args, **kwargs) -> Self:
for name, expression in kwargs.items():
self._check_expression_recursive(expression, context=f"alias '{name}'")
return super().alias(*args, **kwargs)
def filter(self, *args, **kwargs) -> Self:
for arg in args:
self._check_expression_recursive(arg, context="filter Q object")
for name, expression in kwargs.items():
self._check_expression_recursive(expression, context=f"filter '{name}'")
return super().filter(*args, **kwargs)
def _check_expression_recursive(self, expression, context="expression"):
"""
式を再帰的にチェックし、ExpressionWrapper でラップされていない四則演算を検出する。
Args:
expression: チェック対象の式
context: エラーメッセージに含めるコンテキスト情報
"""
# 四則演算のチェック
if self._is_unwrapped_arithmetic(expression):
raise ValueError(
f"Arithmetic expression in {context} must be wrapped with ExpressionWrapper "
f"to explicitly define output_field. "
f"Example: ExpressionWrapper(F('field1') + F('field2'), output_field=BigIntegerField())"
)
# Aggregate の再帰チェック
if isinstance(expression, Aggregate):
# Aggregate の source_expressions をチェック
if hasattr(expression, "source_expressions"):
for source_expr in expression.source_expressions:
self._check_expression_recursive(source_expr, context=f"{context} > Aggregate source")
# Aggregate の filter をチェック
if hasattr(expression, "filter") and expression.filter is not None:
self._check_expression_recursive(expression.filter, context=f"{context} > Aggregate filter")
# Case の再帰チェック
elif isinstance(expression, Case):
# Case の cases(When のリスト)をチェック
if hasattr(expression, "cases"):
for when in expression.cases:
if isinstance(when, When):
# condition をチェック
if hasattr(when, "condition"):
self._check_expression_recursive(when.condition, context=f"{context} > Case When condition")
# then (result) をチェック
if hasattr(when, "result"):
self._check_expression_recursive(when.result, context=f"{context} > Case When then")
# Case の default をチェック
if hasattr(expression, "default") and expression.default is not None:
self._check_expression_recursive(expression.default, context=f"{context} > Case default")
# Q オブジェクトの再帰チェック
elif isinstance(expression, models.Q):
for child in expression.children:
if isinstance(child, models.Q):
self._check_expression_recursive(child, context=f"{context} > Q")
elif isinstance(child, tuple) and len(child) == 2:
key, value = child
self._check_expression_recursive(value, context=f"{context} > Q '{key}'")
# その他の Expression の source_expressions をチェック
elif hasattr(expression, "source_expressions"):
for source_expr in expression.source_expressions:
self._check_expression_recursive(source_expr, context=context)
def _is_unwrapped_arithmetic(self, expression):
"""
与えられた expression が ExpressionWrapper でラップされていない四則演算かどうかをチェック。
"""
# ExpressionWrapper でラップされている場合は問題ない
if isinstance(expression, ExpressionWrapper):
return False
# CombinedExpression(四則演算)の場合はエラー対象
if isinstance(expression, CombinedExpression):
return True
return False
テストコード
from django.contrib.contenttypes.models import ContentType
from django.db.models import (
Avg,
Case,
CharField,
Count,
ExpressionWrapper,
F,
IntegerField,
Q,
Sum,
Value,
When,
)
from django.db.models.functions import Coalesce
from django.test import TestCase
from base_queryset import BaseQuerySet # BaseQuerySetが記載されている箇所に変更してください。
class BaseQuerySetTestCase(TestCase):
def test_multiple_arithmetic_operations(self):
"""
複数の四則演算(+, -, *, /)すべてでチェックが機能することを確認
"""
operations = [
("add", F("id") + Value(1)),
("subtract", F("id") - Value(1)),
("multiply", F("id") * Value(2)),
("divide", F("id") / Value(2)),
("add_field", F("id") + F("id")),
("subtract_field", F("id") - F("id")),
("multiply_field", F("id") * F("id")),
("divide_field", F("id") / F("id")),
("add_value", Value(1) + Value(1)),
("subtract_value", Value(1) - Value(1)),
("multiply_value", Value(1) * Value(1)),
("divide_value", Value(1) / Value(1)),
("add_function", Coalesce(F("id"), Value(1)) + Coalesce(F("id"), Value(1))),
("subtract_function", Coalesce(F("id"), Value(1)) - Coalesce(F("id"), Value(1))),
("multiply_function", Coalesce(F("id"), Value(1)) * Coalesce(F("id"), Value(1))),
("divide_function", Coalesce(F("id"), Value(1)) / Coalesce(F("id"), Value(1))),
]
for name, expression in operations:
with self.subTest(operation=name):
# annotate でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(**{name: expression}).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn(name, str(context.exception))
# annotate で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.annotate(**{name: ExpressionWrapper(expression, output_field=IntegerField())})
.first()
)
self.assertIsNotNone(result)
self.assertTrue(hasattr(result, name))
# alias でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).alias(**{name: expression}).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn(name, str(context.exception))
# alias で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.alias(**{name: ExpressionWrapper(expression, output_field=IntegerField())})
.first()
)
self.assertIsNotNone(result)
# filter でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).filter(id=expression).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
# filter で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.filter(id=ExpressionWrapper(expression, output_field=IntegerField()))
.first()
)
# filter Qオブジェクトでエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).filter(Q(id=expression)).first()
def test_aggregate_filter_arithmetic_operations(self):
"""
Aggregate系(Sum、Count、Avgなど)のfilterパラメータ内の四則演算をチェック
"""
operations = [
("add", F("id") + Value(1)),
("subtract", F("id") - Value(1)),
("multiply", F("id") * Value(2)),
("divide", F("id") / Value(2)),
]
aggregates = [
("Sum", Sum),
("Count", Count),
("Avg", Avg),
]
for agg_name, agg_class in aggregates:
for op_name, expression in operations:
with self.subTest(aggregate=agg_name, operation=op_name):
# Aggregate の filter パラメータでエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
test_agg=agg_class("id", filter=Q(id=expression))
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn("Aggregate", str(context.exception))
# Aggregate の filter パラメータで ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.annotate(
test_agg=agg_class(
"id",
filter=Q(id=ExpressionWrapper(expression, output_field=IntegerField())),
)
)
.first()
)
self.assertIsNotNone(result)
self.assertTrue(hasattr(result, "test_agg"))
def test_case_when_arithmetic_operations(self):
"""
Case/When の condition パラメータ内の四則演算をチェック
"""
operations = [
("add", F("id") + Value(1)),
("subtract", F("id") - Value(1)),
("multiply", F("id") * Value(2)),
("divide", F("id") / Value(2)),
]
for op_name, expression in operations:
with self.subTest(operation=op_name):
# Case/When の condition でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
test_case=Case(
When(condition=Q(id=expression), then=Value("A")),
default=Value("B"),
output_field=CharField(),
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn("Case When condition", str(context.exception))
# Case/When の condition で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.annotate(
test_case=Case(
When(
condition=Q(id=ExpressionWrapper(expression, output_field=IntegerField())),
then=Value("A"),
),
default=Value("B"),
output_field=CharField(),
)
)
.first()
)
self.assertIsNotNone(result)
self.assertTrue(hasattr(result, "test_case"))
# Case/When の then でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
test_case=Case(
When(condition=Q(id__gt=0), then=expression),
default=Value(0),
output_field=IntegerField(),
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn("Case When then", str(context.exception))
# Case/When の then で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.annotate(
test_case=Case(
When(
condition=Q(id__gt=0),
then=ExpressionWrapper(expression, output_field=IntegerField()),
),
default=Value(0),
output_field=IntegerField(),
)
)
.first()
)
self.assertIsNotNone(result)
self.assertTrue(hasattr(result, "test_case"))
# Case の default でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
test_case=Case(
When(condition=Q(id__gt=0), then=Value(1)),
default=expression,
output_field=IntegerField(),
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
self.assertIn("Case default", str(context.exception))
# Case の default で ExpressionWrapper でラップした場合は正常に動作
result = (
BaseQuerySet(model=ContentType)
.annotate(
test_case=Case(
When(condition=Q(id__gt=0), then=Value(1)),
default=ExpressionWrapper(expression, output_field=IntegerField()),
output_field=IntegerField(),
)
)
.first()
)
self.assertIsNotNone(result)
self.assertTrue(hasattr(result, "test_case"))
def test_nested_expressions(self):
"""
ネストされた式(Aggregate内のCase、Case内のAggregate等)の四則演算をチェック
"""
expression = F("id") + Value(1)
# alias 内の Aggregate でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).alias(total=Sum("id", filter=Q(id=expression))).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
# alias 内の Case でエラーになることを確認
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).alias(
status=Case(
When(condition=Q(id=expression), then=Value("A")), default=Value("B"), output_field=CharField()
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
# Aggregate 内の Case でエラーになることを確認(then に四則演算)
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
total=Sum(
Case(When(condition=Q(id__gt=0), then=expression), default=Value(0), output_field=IntegerField())
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
# Case 内の Aggregate でエラーになることを確認(Aggregate の filter に四則演算)
with self.assertRaises(ValueError) as context:
BaseQuerySet(model=ContentType).annotate(
result=Case(
When(condition=Q(id__gt=0), then=Sum("id", filter=Q(id=expression))),
default=Value(0),
output_field=IntegerField(),
)
).first()
self.assertIn("must be wrapped with ExpressionWrapper", str(context.exception))
以上です。
さいごに
silently fail 自体を消すことはできませんが、意図せずオーバーフローする可能性自体を下げることができること、明示的に書くことで型指定がおかしいということにも気づきやすいことから、弊社では上記の対処方法をこの問題に対応することにしました。
参考になれば幸いです。
Discussion