bubusuke です。
Djangoのget()を使っていて、「エラーになったけど、原因が追いにくい…」「どんな条件で検索してたっけ？」と感じたことはありませんか？
{モデル名} matching query does not exist.
こんなエラーはしんどい。。そんな「調査しにくさ」を解消するために、私たちは独自QuerySetでgetを拡張しています。
前回の記事でアプリケーション全体に独自QuerySetを導入する話に触れました。
今回はその続きとして、getメソッドの拡張内容と、独自QuerySet設定漏れを防ぐためのテストをどのように組み込んでいるかを紹介します。
getメソッドの拡張内容
- Django標準の
getは、エラー時に引数情報が出ないため、調査しやすいように条件を含めるようにしています。from eで例外を連鎖させて、スタックトレースを残すようにケアしてます。 -
SubQueryやQuerySetを指定した場合にも対応できるようにしています。 - 機密情報を扱う場合は値をマスクするようにしましょう（弊社では今のところ該当ケースがないため未対応です）。
from django.core.exceptions import ValidationError
from django.db import models
class BaseQuerySet(models.QuerySet):
#########################################################
# getメソッドのオーバーライド
#########################################################
def get(self, *args, **kwargs):
"""
エラーメッセージにargsとkwargsを追加することで、エラーの原因を特定しやすくする。
"""
try:
return super().get(*args, **kwargs)
except self.model.DoesNotExist as e:
_args, _kwargs = self._str_args(args, kwargs)
raise self.model.DoesNotExist(
f"{self.model._meta.object_name} matching query does not exist. args: {_args}, kwargs: {_kwargs}"
) from e
except self.model.MultipleObjectsReturned as e:
_args, _kwargs = self._str_args(args, kwargs)
raise self.model.MultipleObjectsReturned(
f"{self.model._meta.object_name} matching query returns multiple objects. args: {_args}, kwargs: {_kwargs}"
) from e
except ValidationError as e:
_args, _kwargs = self._str_args(args, kwargs)
raise ValidationError(
f"Failed to get {self.model._meta.object_name}. args: {_args}, kwargs: {_kwargs}. Error: {e.message}"
) from e
except Exception as e:
_args, _kwargs = self._str_args(args, kwargs)
raise Exception(
f"Failed to get {self.model._meta.object_name}. args: {_args}, kwargs: {_kwargs}. Error: {e}"
) from e
def _str_args(self, args, kwargs):
_args = []
_kwargs = {}
for arg in args:
_args.append(self._format_value(arg))
for k, v in kwargs.items():
_kwargs[k] = self._format_value(v)
return _args, _kwargs
def _format_value(self, v):
if isinstance(v, models.Model):
return f"{v._meta.object_name}(id: {v.id})"
elif isinstance(v, models.Q):
return f" {v.connector} (".join([self._format_value(child) for child in v.children]) + ")"
elif isinstance(v, models.Subquery):
return str(v.query)
elif isinstance(v, models.QuerySet):
return str(v.query)
elif isinstance(v, list):
return [self._format_value(item) for item in v]
else:
return str(v)
#########################################################
# ExpressionWrapperなしで四則演算することを禁止するためのオーバーライド
#########################################################
# 前回の記事で紹介した内容なので割愛
実際に試してみましょう。
以下のようなコードを実行すると。。。
from django.contrib.contenttypes.models import ContentType
from django.db.models import Exists, Q
from django.test import TestCase
from faker import Faker
from .base_queryset import BaseQuerySet
fake = Faker()
class BaseQuerySetGetTestCase(TestCase):
def test_get(self):
# app1, app2, model2 は各自実際に定義しているモデル名、apps名に書き換えてください。
try:
BaseQuerySet(model=ContentType).get(
Q(id=fake.random_int(min=10000, max=1000000))
& Exists(ContentType.objects.filter(id=fake.random_int(min=1, max=1000000)))
| Exists(ContentType.objects.filter(id=fake.random_int(min=1, max=1000000))),
app_label="app1",
app_label__in=["app1", "app2"],
id=ContentType.objects.get(app_label="app2", model="model2").id,
model=ContentType.objects.filter(app_label="app2", model="model2").first(),
)
except Exception as e:
print(e)
self.assertIn("args:", str(e))
self.assertIn("kwargs:", str(e))
無事、エラーに検索条件が含まれていることを確認できました。
ContentType matching query does not exist. args: ['(\'id\', 95733) AND (SELECT 1 AS "a" FROM "django_content_type" WHERE "django_content_type"."id" = 969938 LIMIT 1) OR (SELECT 1 AS "a" FROM "django_content_type" WHERE "django_content_type"."id" = 555844 LIMIT 1)'], kwargs: {'app_label': 'app1', 'app_label__in': ['app1', 'app2'], 'id': '311', 'model': 'ContentType(id: 311)'}
拡張前のエラーメッセージと比べると、かなり情報が増えたことがわかります。
ContentType matching query does not exist.
独自のQuerySetの設定漏れを防ぐ仕組み
- 設定漏れのモデルのテストを検知するテストを追加します。
- このテストをCIで実行することで設定漏れを防ぎます（CIへの導入方法は割愛します）。
from django.apps import apps
from django.test import TestCase
from bns.utils.base_queryset import BaseQuerySet
class BaseQuerySetCoverageTestCase(TestCase):
def test_base_queryset_coverage(self):
"""
model 定義をすべてチェックして想定外の設定漏れがないか確認する
"""
# model 単位でチェック不要なもの
ignored_models = []
ngs = []
for app in apps.get_app_configs():
if "site-packages" in app.path:
# サードパーティのモデルは無視
continue
# model 単位でチェック
for model in app.models.values():
if model._meta.auto_created:
# 自動生成されたモデルは無視(m2m経由の中間モデルなど)
continue
# BaseQuerySetを継承しているかチェック
qs = model.objects.get_queryset()
if not isinstance(qs, BaseQuerySet):
name = model._meta.model_name
full_name = f"{app.label}.{name}"
if full_name in ignored_models:
continue
ngs.append(full_name)
# NOTE:
# - ここでエラーが発生した場合は、BaseQuerySetの設定漏れ
# - 設定不要であればignoreを追加すること
self.assertEqual(0, len(ngs), f"以下のモデルのクエリセットがBaseQuerySetを継承していません: {ngs}")
さいごに
getの例外に検索条件が載るだけで、再現や原因特定の往復が一気に減ります。さらに、独自QuerySetの設定漏れをCIで拾う仕組みを入れておけば、時間が経っても品質を保てます。
運用に入れる際は、以下の2点だけ気をつけると安全です。
- 機密値の扱い：ID以外の個人情報やトークンなどがエラーメッセージに出ないよう、必要に応じてマスキングを入れる。
- メッセージの長さ：巨大なSubqueryや複雑なQでメッセージが長くなりすぎる場合は、適度にトリミングする。
例外メッセージの工夫などの小さな改善の積み重ねがデバッグ時間を確実に短くします。今日の一歩で、運用をもっと楽にしていきましょう。
関連記事
Discussion