add ObjectIdField

Author: WaVEV

django_mongodb/fields/__init__.py

@@ -1,8 +1,9 @@
 from .auto import ObjectIdAutoField
 from .duration import register_duration_field
 from .json import register_json_field
+from .objectid import ObjectIdField
 
-__all__ = ["register_fields", "ObjectIdAutoField"]
+__all__ = ["register_fields", "ObjectIdAutoField", "ObjectIdField"]
 
 
 def register_fields():

django_mongodb/fields/auto.py

@@ -2,15 +2,11 @@
 from django.core import exceptions
 from django.db.models.fields import AutoField
 from django.utils.functional import cached_property
-from django.utils.translation import gettext_lazy as _
 
+from .objectid import ObjectIdMixin
 
-class ObjectIdAutoField(AutoField):
-    default_error_messages = {
-        "invalid": _("“%(value)s” value must be an Object Id."),
-    }
-    description = _("Object Id")
 
+class ObjectIdAutoField(ObjectIdMixin, AutoField):
     def __init__(self, *args, **kwargs):
         kwargs["db_column"] = "_id"
         super().__init__(*args, **kwargs)
@@ -42,12 +38,6 @@ def get_prep_value(self, value):
     def get_internal_type(self):
         return "ObjectIdAutoField"
 
-    def db_type(self, connection):
-        return "objectId"
-
-    def rel_db_type(self, connection):
-        return "objectId"
-
     def to_python(self, value):
         if value is None or isinstance(value, int):
             return value

django_mongodb/fields/objectid.py

@@ -0,0 +1,48 @@
+from bson import ObjectId, errors
+from django.core import exceptions
+from django.db.models.fields import Field
+from django.utils.translation import gettext_lazy as _
+
+
+class ObjectIdMixin:
+    default_error_messages = {
+        "invalid": _("“%(value)s” value must be an Object Id."),
+    }
+    description = _("Object Id")
+
+    def db_type(self, connection):
+        return "objectId"
+
+    def rel_db_type(self, connection):
+        return "objectId"
+
+    def get_prep_value(self, value):
+        if value is None or isinstance(value, ObjectId):
+            return value
+        try:
+            return ObjectId(value)
+        except (errors.InvalidId, TypeError) as e:
+            raise ValueError(f"Field '{self.name}' expected an ObjectId but got {value!r}.") from e
+
+    def to_python(self, value):
+        if value is None:
+            return value
+        try:
+            return ObjectId(value)
+        except (errors.InvalidId, TypeError):
+            raise exceptions.ValidationError(
+                self.error_messages["invalid"],
+                code="invalid",
+                params={"value": value},
+            ) from None
+
+
+class ObjectIdField(ObjectIdMixin, Field):
+    def deconstruct(self):
+        name, path, args, kwargs = super().deconstruct()
+        if path.startswith("django_mongodb.fields.objectid"):
+            path = path.replace("django_mongodb.fields.objectid", "django_mongodb.fields")
+        return name, path, args, kwargs
+
+    def get_internal_type(self):
+        return "ObjectIdField"

docs/source/fields.rst

@@ -0,0 +1,13 @@
+Model field reference
+=====================
+
+.. module:: django_mongodb.fields
+
+Some MongoDB-specific fields are available in ``django_mongodb.fields``.
+
+``ObjectIdField``
+-----------------
+
+.. class:: ObjectIdField
+
+Stores an :class:`~bson.objectid.ObjectId`.

docs/source/index.rst

@@ -5,6 +5,7 @@ django-mongodb 5.0.x documentation
    :maxdepth: 1
    :caption: Contents:
 
+   fields
    querysets
 
 Indices and tables

tests/model_fields_/test_objectidfield.py

@@ -0,0 +1,57 @@
+from bson import ObjectId
+from django.core import exceptions
+from django.test import SimpleTestCase
+
+from django_mongodb.fields import ObjectIdField
+
+
+class MethodTests(SimpleTestCase):
+    def test_deconstruct(self):
+        field = ObjectIdField()
+        name, path, args, kwargs = field.deconstruct()
+        self.assertEqual(path, "django_mongodb.fields.ObjectIdField")
+        self.assertEqual(args, [])
+        self.assertEqual(kwargs, {})
+
+    def test_get_internal_type(self):
+        f = ObjectIdField()
+        self.assertEqual(f.get_internal_type(), "ObjectIdField")
+
+    def test_to_python_string(self):
+        value = "1" * 24
+        self.assertEqual(ObjectIdField().to_python(value), ObjectId(value))
+
+    def test_to_python_objectid(self):
+        value = ObjectId("1" * 24)
+        self.assertEqual(ObjectIdField().to_python(value), value)
+
+    def test_to_python_null(self):
+        self.assertIsNone(ObjectIdField().to_python(None))
+
+    def test_to_python_invalid_value(self):
+        f = ObjectIdField()
+        for invalid_value in ["None", {}, [], 123]:
+            with self.subTest(invalid_value=invalid_value):
+                msg = f"['“{invalid_value}” value must be an Object Id.']"
+                with self.assertRaisesMessage(exceptions.ValidationError, msg):
+                    f.to_python(invalid_value)
+
+    def test_get_prep_value_string(self):
+        value = "1" * 24
+        self.assertEqual(ObjectIdField().get_prep_value(value), ObjectId(value))
+
+    def test_get_prep_value_objectid(self):
+        value = ObjectId("1" * 24)
+        self.assertEqual(ObjectIdField().get_prep_value(value), value)
+
+    def test_get_prep_value_null(self):
+        self.assertIsNone(ObjectIdField().get_prep_value(None))
+
+    def test_get_prep_value_invalid_values(self):
+        f = ObjectIdField()
+        f.name = "test"
+        for invalid_value in ["None", {}, [], 123]:
+            with self.subTest(invalid_value=invalid_value):
+                msg = f"Field 'test' expected an ObjectId but got {invalid_value!r}."
+                with self.assertRaisesMessage(ValueError, msg):
+                    f.get_prep_value(invalid_value)

tests/queries_/models.py

@@ -1,5 +1,7 @@
 from django.db import models
 
+from django_mongodb.fields import ObjectIdAutoField, ObjectIdField
+
 
 class Author(models.Model):
     name = models.CharField(max_length=10)
@@ -14,3 +16,40 @@ class Book(models.Model):
 
     def __str__(self):
         return self.title
+
+
+class Tag(models.Model):
+    name = models.CharField(max_length=10)
+    parent = models.ForeignKey(
+        "self",
+        models.SET_NULL,
+        blank=True,
+        null=True,
+        related_name="children",
+    )
+    group_id = ObjectIdField(null=True)
+
+    def __str__(self):
+        return self.name
+
+
+class Order(models.Model):
+    id = ObjectIdAutoField(primary_key=True)
+    name = models.CharField(max_length=12, null=True, default="")
+
+    class Meta:
+        ordering = ("pk",)
+
+    def __str__(self):
+        return str(self.pk)
+
+
+class OrderItem(models.Model):
+    order = models.ForeignKey(Order, models.CASCADE, related_name="items")
+    status = ObjectIdField(null=True)
+
+    class Meta:
+        ordering = ("pk",)
+
+    def __str__(self):
+        return str(self.pk)

tests/queries_/test_objectid.py

@@ -0,0 +1,111 @@
+from bson import ObjectId
+from django.test import TestCase
+
+from .models import Order, OrderItem, Tag
+
+
+class ObjectIdTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.group_id_str_1 = "1" * 24
+        cls.group_id_obj_1 = ObjectId(cls.group_id_str_1)
+        cls.group_id_str_2 = "2" * 24
+        cls.group_id_obj_2 = ObjectId(cls.group_id_str_2)
+
+        cls.t1 = Tag.objects.create(name="t1")
+        cls.t2 = Tag.objects.create(name="t2", parent=cls.t1)
+        cls.t3 = Tag.objects.create(name="t3", parent=cls.t1, group_id=cls.group_id_str_1)
+        cls.t4 = Tag.objects.create(name="t4", parent=cls.t3, group_id=cls.group_id_obj_2)
+        cls.t5 = Tag.objects.create(name="t5", parent=cls.t3)
+
+    def test_filter_group_id_is_null_false(self):
+        """Filter objects where group_id is not null."""
+        qs = Tag.objects.filter(group_id__isnull=False).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3, self.t4])
+
+    def test_filter_group_id_is_null_true(self):
+        """Filter objects where group_id is null."""
+        qs = Tag.objects.filter(group_id__isnull=True).order_by("name")
+        self.assertSequenceEqual(qs, [self.t1, self.t2, self.t5])
+
+    def test_filter_group_id_equal_str(self):
+        """Filter by group_id with a specific string value."""
+        qs = Tag.objects.filter(group_id=self.group_id_str_1).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3])
+
+    def test_filter_group_id_equal_obj(self):
+        """Filter by group_id with a specific ObjectId value."""
+        qs = Tag.objects.filter(group_id=self.group_id_obj_1).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3])
+
+    def test_filter_group_id_in_str_values(self):
+        """Filter by group_id with string values in a list."""
+        ids = [self.group_id_str_1, self.group_id_str_2]
+        qs = Tag.objects.filter(group_id__in=ids).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3, self.t4])
+
+    def test_filter_group_id_in_obj_values(self):
+        """Filter by group_id with ObjectId values in a list."""
+        ids = [self.group_id_obj_1, self.group_id_obj_2]
+        qs = Tag.objects.filter(group_id__in=ids).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3, self.t4])
+
+    def test_filter_group_id_equal_subquery(self):
+        """Filter by group_id using a subquery."""
+        subquery = Tag.objects.filter(name="t3").values("group_id")
+        qs = Tag.objects.filter(group_id__in=subquery).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3])
+
+    def test_filter_group_id_in_subquery(self):
+        """Filter by group_id using a subquery with multiple values."""
+        subquery = Tag.objects.filter(name__in=["t3", "t4"]).values("group_id")
+        qs = Tag.objects.filter(group_id__in=subquery).order_by("name")
+        self.assertSequenceEqual(qs, [self.t3, self.t4])
+
+    def test_filter_parent_by_children_values_str(self):
+        """Query to select parents of children with specific string group_id."""
+        child_ids = Tag.objects.filter(group_id=self.group_id_str_1).values_list("id", flat=True)
+        parent_qs = Tag.objects.filter(children__id__in=child_ids).distinct().order_by("name")
+        self.assertSequenceEqual(parent_qs, [self.t1])
+
+    def test_filter_parent_by_children_values_obj(self):
+        """Query to select parents of children with specific ObjectId group_id."""
+        child_ids = Tag.objects.filter(group_id=self.group_id_obj_1).values_list("id", flat=True)
+        parent_qs = Tag.objects.filter(children__id__in=child_ids).distinct().order_by("name")
+        self.assertSequenceEqual(parent_qs, [self.t1])
+
+    def test_filter_group_id_union_with_str(self):
+        """Combine queries using union with string values."""
+        qs_a = Tag.objects.filter(group_id=self.group_id_str_1)
+        qs_b = Tag.objects.filter(group_id=self.group_id_str_2)
+        union_qs = qs_a.union(qs_b).order_by("name")
+        self.assertSequenceEqual(union_qs, [self.t3, self.t4])
+
+    def test_filter_group_id_union_with_obj(self):
+        """Combine queries using union with ObjectId values."""
+        qs_a = Tag.objects.filter(group_id=self.group_id_obj_1)
+        qs_b = Tag.objects.filter(group_id=self.group_id_obj_2)
+        union_qs = qs_a.union(qs_b).order_by("name")
+        self.assertSequenceEqual(union_qs, [self.t3, self.t4])
+
+    def test_filter_invalid_object_id(self):
+        value = "value1"
+        msg = f"Field 'group_id' expected an ObjectId but got '{value}'."
+        with self.assertRaisesMessage(ValueError, msg):
+            Tag.objects.filter(group_id=value)
+
+    def test_values_in_subquery(self):
+        # If a values() queryset is used, then the given values will be used
+        # instead of forcing use of the relation's field.
+        o1 = Order.objects.create()
+        o2 = Order.objects.create()
+        oi1 = OrderItem.objects.create(order=o1, status=None)
+        oi1.status = oi1.pk
+        oi1.save()
+        OrderItem.objects.create(order=o2, status=None)
+        # The query below should match o1 as it has related order_item with
+        # id == status.
+        self.assertSequenceEqual(
+            Order.objects.filter(items__in=OrderItem.objects.values_list("status")),
+            [o1],
+        )