16950450531

Committed 13 Aug 2025 10:07PM UTC coverage: 94.028% (-0.5%) from 94.481%

Build # 16950450531

Build Type

push

github

Committed by

bagerard

Commit Message

add useful cr comment for .path in install_mongo

Run Details

5322 of 5660 relevant lines covered (94.03%)

1.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.19

/mongoengine/base/fields.py

import contextlib
import operator
import threading
import weakref

import pymongo
from bson import SON, DBRef, ObjectId

from mongoengine.base.common import UPDATE_OPERATORS
from mongoengine.base.datastructures import (
    BaseDict,
    BaseList,
    EmbeddedDocumentList,
)
from mongoengine.common import _import_class
from mongoengine.errors import DeprecatedError, ValidationError

__all__ = ("BaseField", "ComplexBaseField", "ObjectIdField", "GeoJsonBaseField")


@contextlib.contextmanager
def _no_dereference_for_fields(*fields):
    """Context manager for temporarily disabling a Field's auto-dereferencing
    (meant to be used from no_dereference context manager)"""
    try:
        for field in fields:
            field._incr_no_dereference_context()
        yield None
    finally:
        for field in fields:
            field._decr_no_dereference_context()


class BaseField:
    """A base class for fields in a MongoDB document. Instances of this class
    may be added to subclasses of `Document` to define a document's schema.
    """

    name = None  # set in TopLevelDocumentMetaclass
    _geo_index = False
    _auto_gen = False  # Call `generate` to generate a value
    _thread_local_storage = threading.local()

    # These track each time a Field instance is created. Used to retain order.
    # The auto_creation_counter is used for fields that MongoEngine implicitly
    # creates, creation_counter is used for all user-specified fields.
    creation_counter = 0
    auto_creation_counter = -1

    def __init__(
        self,
        db_field=None,
        required=False,
        default=None,
        unique=False,
        unique_with=None,
        primary_key=False,
        validation=None,
        choices=None,
        null=False,
        sparse=False,
        **kwargs,
    ):
        """
        :param db_field: The database field to store this field in
            (defaults to the name of the field)
        :param required: If the field is required. Whether it has to have a
            value or not. Defaults to False.
        :param default: (optional) The default value for this field if no value
            has been set, if the value is set to None or has been unset. It can be a
            callable.
        :param unique: Is the field value unique or not (Creates an index).  Defaults to False.
        :param unique_with: (optional) The other field this field should be
            unique with (Creates an index).
        :param primary_key: Mark this field as the primary key ((Creates an index)). Defaults to False.
        :param validation: (optional) A callable to validate the value of the
            field. The callable takes the value as parameter and should raise
            a ValidationError if validation fails
        :param choices: (optional) The valid choices
        :param null: (optional) If the field value can be null when a default exists. If not set, the default value
            will be used in case a field with a default value is set to None. Defaults to False.
        :param sparse: (optional) `sparse=True` combined with `unique=True` and `required=False`
            means that uniqueness won't be enforced for `None` values (Creates an index). Defaults to False.
        :param **kwargs: (optional) Arbitrary indirection-free metadata for
            this field can be supplied as additional keyword arguments and
            accessed as attributes of the field. Must not conflict with any
            existing attributes. Common metadata includes `verbose_name` and
            `help_text`.
        """
        self.db_field = db_field if not primary_key else "_id"

        self.required = required or primary_key
        self.default = default
        self.unique = bool(unique or unique_with)
        self.unique_with = unique_with
        self.primary_key = primary_key
        self.validation = validation
        self.choices = choices
        self.null = null
        self.sparse = sparse
        self._owner_document = None

        self.__auto_dereference = True

        # Make sure db_field is a string (if it's explicitly defined).
        if self.db_field is not None and not isinstance(self.db_field, str):
            raise TypeError("db_field should be a string.")

        # Make sure db_field doesn't contain any forbidden characters.
        if isinstance(self.db_field, str) and (
            "." in self.db_field
            or "\0" in self.db_field
            or self.db_field.startswith("$")
        ):
            raise ValueError(
                'field names cannot contain dots (".") or null characters '
                '("\\0"), and they must not start with a dollar sign ("$").'
            )

        # Detect and report conflicts between metadata and base properties.
        conflicts = set(dir(self)) & set(kwargs)
        if conflicts:
            raise TypeError(
                "%s already has attribute(s): %s"
                % (self.__class__.__name__, ", ".join(conflicts))
            )

        # Assign metadata to the instance
        # This efficient method is available because no __slots__ are defined.
        self.__dict__.update(kwargs)

        # Adjust the appropriate creation counter, and save our local copy.
        if self.db_field == "_id":
            self.creation_counter = BaseField.auto_creation_counter
            BaseField.auto_creation_counter -= 1
        else:
            self.creation_counter = BaseField.creation_counter
            BaseField.creation_counter += 1

    def set_auto_dereferencing(self, value):
        self.__auto_dereference = value

    @property
    def _no_dereference_context_local(self):
        if not hasattr(self._thread_local_storage, "no_dereference_context"):
            self._thread_local_storage.no_dereference_context = 0
        return self._thread_local_storage.no_dereference_context

    @property
    def _no_dereference_context_is_set(self):
        return self._no_dereference_context_local > 0

    def _incr_no_dereference_context(self):
        self._thread_local_storage.no_dereference_context = (
            self._no_dereference_context_local + 1
        )

    def _decr_no_dereference_context(self):
        self._thread_local_storage.no_dereference_context = (
            self._no_dereference_context_local - 1
        )

    @property
    def _auto_dereference(self):
        return self.__auto_dereference and not self._no_dereference_context_is_set

    def __get__(self, instance, owner):
        """Descriptor for retrieving a value from a field in a document."""
        if instance is None:
            # Document class being used rather than a document object
            return self

        # Get value from document instance if available
        return instance._data.get(self.name)

    def __set__(self, instance, value):
        """Descriptor for assigning a value to a field in a document."""
        # If setting to None and there is a default value provided for this
        # field, then set the value to the default value.
        if value is None:
            if self.null:
                value = None
            elif self.default is not None:
                value = self.default
                if callable(value):
                    value = value()

        if instance._initialised:
            try:
                value_has_changed = (
                    self.name not in instance._data
                    or instance._data[self.name] != value
                )
                if value_has_changed:
                    instance._mark_as_changed(self.name)
            except Exception:
                # Some values can't be compared and throw an error when we
                # attempt to do so (e.g. tz-naive and tz-aware datetimes).
                # Mark the field as changed in such cases.
                instance._mark_as_changed(self.name)

        EmbeddedDocument = _import_class("EmbeddedDocument")
        if isinstance(value, EmbeddedDocument):
            value._instance = weakref.proxy(instance)
        elif isinstance(value, (list, tuple)):
            for v in value:
                if isinstance(v, EmbeddedDocument):
                    v._instance = weakref.proxy(instance)

        instance._data[self.name] = value

    def error(self, message="", errors=None, field_name=None):
        """Raise a ValidationError."""
        field_name = field_name if field_name else self.name
        raise ValidationError(message, errors=errors, field_name=field_name)

    def to_python(self, value):
        """Convert a MongoDB-compatible type to a Python type."""
        return value

    def to_mongo(self, value):
        """Convert a Python type to a MongoDB-compatible type."""
        return self.to_python(value)

    def _to_mongo_safe_call(self, value, use_db_field=True, fields=None):
        """Helper method to call to_mongo with proper inputs."""
        f_inputs = self.to_mongo.__code__.co_varnames
        ex_vars = {}
        if "fields" in f_inputs:
            ex_vars["fields"] = fields

        if "use_db_field" in f_inputs:
            ex_vars["use_db_field"] = use_db_field

        return self.to_mongo(value, **ex_vars)

    def prepare_query_value(self, op, value):
        """Prepare a value that is being used in a query for PyMongo."""
        if op in UPDATE_OPERATORS:
            self.validate(value)
        return value

    def validate(self, value, clean=True):
        """Perform validation on a value."""
        pass

    def _validate_choices(self, value):
        Document = _import_class("Document")
        EmbeddedDocument = _import_class("EmbeddedDocument")

        choice_list = self.choices
        if isinstance(next(iter(choice_list)), (list, tuple)):
            # next(iter) is useful for sets
            choice_list = [k for k, _ in choice_list]

        # Choices which are other types of Documents
        if isinstance(value, (Document, EmbeddedDocument)):
            if not any(isinstance(value, c) for c in choice_list):
                self.error("Value must be an instance of %s" % (choice_list))
        # Choices which are types other than Documents
        else:
            values = value if isinstance(value, (list, tuple)) else [value]
            if len(set(values) - set(choice_list)):
                self.error("Value must be one of %s" % str(choice_list))

    def _validate(self, value, **kwargs):
        # Check the Choices Constraint
        if self.choices:
            self._validate_choices(value)

        # check validation argument
        if self.validation is not None:
            if callable(self.validation):
                try:
                    # breaking change of 0.18
                    # Get rid of True/False-type return for the validation method
                    # in favor of having validation raising a ValidationError
                    ret = self.validation(value)
                    if ret is not None:
                        raise DeprecatedError(
                            "validation argument for `%s` must not return anything, "
                            "it should raise a ValidationError if validation fails"
                            % self.name
                        )
                except ValidationError as ex:
                    self.error(str(ex))
            else:
                raise ValueError(
                    'validation argument for `"%s"` must be a ' "callable." % self.name
                )

        self.validate(value, **kwargs)

    @property
    def owner_document(self):
        return self._owner_document

    def _set_owner_document(self, owner_document):
        self._owner_document = owner_document

    @owner_document.setter
    def owner_document(self, owner_document):
        self._set_owner_document(owner_document)


class ComplexBaseField(BaseField):
    """Handles complex fields, such as lists / dictionaries.

    Allows for nesting of embedded documents inside complex types.
    Handles the lazy dereferencing of a queryset by lazily dereferencing all
    items in a list / dict rather than one at a time.
    """

    def __init__(self, field=None, **kwargs):
        if field is not None and not isinstance(field, BaseField):
            raise TypeError(
                f"field argument must be a Field instance (e.g {self.__class__.__name__}(StringField()))"
            )
        self.field = field
        super().__init__(**kwargs)

    @staticmethod
    def _lazy_load_refs(instance, name, ref_values, *, max_depth):
        _dereference = _import_class("DeReference")()
        documents = _dereference(
            ref_values,
            max_depth=max_depth,
            instance=instance,
            name=name,
        )
        return documents

    def __set__(self, instance, value):
        # Some fields e.g EnumField are converted upon __set__
        # So it is fair to mimic the same behavior when using e.g ListField(EnumField)
        EnumField = _import_class("EnumField")
        if self.field and isinstance(self.field, EnumField):
            if isinstance(value, (list, tuple)):
                value = [self.field.to_python(sub_val) for sub_val in value]
            elif isinstance(value, dict):
                value = {key: self.field.to_python(sub) for key, sub in value.items()}

        return super().__set__(instance, value)

    def __get__(self, instance, owner):
        """Descriptor to automatically dereference references."""
        if instance is None:
            # Document class being used rather than a document object
            return self

        ReferenceField = _import_class("ReferenceField")
        GenericReferenceField = _import_class("GenericReferenceField")
        EmbeddedDocumentListField = _import_class("EmbeddedDocumentListField")

        auto_dereference = instance._fields[self.name]._auto_dereference

        dereference = auto_dereference and (
            self.field is None
            or isinstance(self.field, (GenericReferenceField, ReferenceField))
        )

        if (
            instance._initialised
            and dereference
            and instance._data.get(self.name)
            and not getattr(instance._data[self.name], "_dereferenced", False)
        ):
            ref_values = instance._data.get(self.name)
            instance._data[self.name] = self._lazy_load_refs(
                ref_values=ref_values, instance=instance, name=self.name, max_depth=1
            )
            if hasattr(instance._data[self.name], "_dereferenced"):
                instance._data[self.name]._dereferenced = True

        value = super().__get__(instance, owner)

        # Convert lists / values so we can watch for any changes on them
        if isinstance(value, (list, tuple)):
            if issubclass(type(self), EmbeddedDocumentListField) and not isinstance(
                value, EmbeddedDocumentList
            ):
                value = EmbeddedDocumentList(value, instance, self.name)
            elif not isinstance(value, BaseList):
                value = BaseList(value, instance, self.name)
            instance._data[self.name] = value
        elif isinstance(value, dict) and not isinstance(value, BaseDict):
            value = BaseDict(value, instance, self.name)
            instance._data[self.name] = value

        if (
            auto_dereference
            and instance._initialised
            and isinstance(value, (BaseList, BaseDict))
            and not value._dereferenced
        ):
            value = self._lazy_load_refs(
                ref_values=value, instance=instance, name=self.name, max_depth=1
            )
            value._dereferenced = True
            instance._data[self.name] = value

        return value

    def to_python(self, value):
        """Convert a MongoDB-compatible type to a Python type."""
        if isinstance(value, str):
            return value

        if hasattr(value, "to_python"):
            return value.to_python()

        BaseDocument = _import_class("BaseDocument")
        if isinstance(value, BaseDocument):
            # Something is wrong, return the value as it is
            return value

        is_list = False
        if not hasattr(value, "items"):
            try:
                is_list = True
                value = {idx: v for idx, v in enumerate(value)}
            except TypeError:  # Not iterable return the value
                return value

        if self.field:
            self.field.set_auto_dereferencing(self._auto_dereference)
            value_dict = {
                key: self.field.to_python(item) for key, item in value.items()
            }
        else:
            Document = _import_class("Document")
            value_dict = {}
            for k, v in value.items():
                if isinstance(v, Document):
                    # We need the id from the saved object to create the DBRef
                    if v.pk is None:
                        self.error(
                            "You can only reference documents once they"
                            " have been saved to the database"
                        )
                    collection = v._get_collection_name()
                    value_dict[k] = DBRef(collection, v.pk)
                elif hasattr(v, "to_python"):
                    value_dict[k] = v.to_python()
                else:
                    value_dict[k] = self.to_python(v)

        if is_list:  # Convert back to a list
            return [
                v for _, v in sorted(value_dict.items(), key=operator.itemgetter(0))
            ]
        return value_dict

    def to_mongo(self, value, use_db_field=True, fields=None):
        """Convert a Python type to a MongoDB-compatible type."""
        Document = _import_class("Document")
        EmbeddedDocument = _import_class("EmbeddedDocument")
        GenericReferenceField = _import_class("GenericReferenceField")

        if isinstance(value, str):
            return value

        if hasattr(value, "to_mongo"):
            if isinstance(value, Document):
                return GenericReferenceField().to_mongo(value)
            cls = value.__class__
            val = value.to_mongo(use_db_field, fields)
            # If it's a document that is not inherited add _cls
            if isinstance(value, EmbeddedDocument):
                val["_cls"] = cls.__name__
            return val

        is_list = False
        if not hasattr(value, "items"):
            try:
                is_list = True
                value = {k: v for k, v in enumerate(value)}
            except TypeError:  # Not iterable return the value
                return value

        if self.field:
            value_dict = {
                key: self.field._to_mongo_safe_call(item, use_db_field, fields)
                for key, item in value.items()
            }
        else:
            value_dict = {}
            for k, v in value.items():
                if isinstance(v, Document):
                    # We need the id from the saved object to create the DBRef
                    if v.pk is None:
                        self.error(
                            "You can only reference documents once they"
                            " have been saved to the database"
                        )

                    # If it's a document that is not inheritable it won't have
                    # any _cls data so make it a generic reference allows
                    # us to dereference
                    meta = getattr(v, "_meta", {})
                    allow_inheritance = meta.get("allow_inheritance")
                    if not allow_inheritance:
                        value_dict[k] = GenericReferenceField().to_mongo(v)
                    else:
                        collection = v._get_collection_name()
                        value_dict[k] = DBRef(collection, v.pk)
                elif hasattr(v, "to_mongo"):
                    cls = v.__class__
                    val = v.to_mongo(use_db_field, fields)
                    # If it's a document that is not inherited add _cls
                    if isinstance(v, (Document, EmbeddedDocument)):
                        val["_cls"] = cls.__name__
                    value_dict[k] = val
                else:
                    value_dict[k] = self.to_mongo(v, use_db_field, fields)

        if is_list:  # Convert back to a list
            return [
                v for _, v in sorted(value_dict.items(), key=operator.itemgetter(0))
            ]
        return value_dict

    def validate(self, value):
        """If field is provided ensure the value is valid."""
        errors = {}
        if self.field:
            if hasattr(value, "items"):
                sequence = value.items()
            else:
                sequence = enumerate(value)
            for k, v in sequence:
                try:
                    self.field._validate(v)
                except ValidationError as error:
                    errors[k] = error.errors or error
                except (ValueError, AssertionError) as error:
                    errors[k] = error

            if errors:
                field_class = self.field.__class__.__name__
                self.error(f"Invalid {field_class} item ({value})", errors=errors)
        # Don't allow empty values if required
        if self.required and not value:
            self.error("Field is required and cannot be empty")

    def prepare_query_value(self, op, value):
        return self.to_mongo(value)

    def lookup_member(self, member_name):
        if self.field:
            return self.field.lookup_member(member_name)
        return None

    def _set_owner_document(self, owner_document):
        if self.field:
            self.field.owner_document = owner_document
        self._owner_document = owner_document


class ObjectIdField(BaseField):
    """A field wrapper around MongoDB's ObjectIds."""

    def to_python(self, value):
        try:
            if not isinstance(value, ObjectId):
                value = ObjectId(value)
        except Exception:
            pass
        return value

    def to_mongo(self, value):
        if isinstance(value, ObjectId):
            return value

        try:
            return ObjectId(str(value))
        except Exception as e:
            self.error(str(e))

    def prepare_query_value(self, op, value):
        if value is None:
            return value
        return self.to_mongo(value)

    def validate(self, value):
        try:
            ObjectId(str(value))
        except Exception:
            self.error("Invalid ObjectID")


class GeoJsonBaseField(BaseField):
    """A geo json field storing a geojson style object."""

    _geo_index = pymongo.GEOSPHERE
    _type = "GeoBase"

    def __init__(self, auto_index=True, *args, **kwargs):
        """
        :param bool auto_index: Automatically create a '2dsphere' index.\
            Defaults to `True`.
        """
        self._name = "%sField" % self._type
        if not auto_index:
            self._geo_index = False
        super().__init__(*args, **kwargs)

    def validate(self, value):
        """Validate the GeoJson object based on its type."""
        if isinstance(value, dict):
            if set(value.keys()) == {"type", "coordinates"}:
                if value["type"] != self._type:
                    self.error(f'{self._name} type must be "{self._type}"')
                return self.validate(value["coordinates"])
            else:
                self.error(
                    "%s can only accept a valid GeoJson dictionary"
                    " or lists of (x, y)" % self._name
                )
                return
        elif not isinstance(value, (list, tuple)):
            self.error("%s can only accept lists of [x, y]" % self._name)
            return

        validate = getattr(self, "_validate_%s" % self._type.lower())
        error = validate(value)
        if error:
            self.error(error)

    def _validate_polygon(self, value, top_level=True):
        if not isinstance(value, (list, tuple)):
            return "Polygons must contain list of linestrings"

        # Quick and dirty validator
        try:
            value[0][0][0]
        except (TypeError, IndexError):
            return "Invalid Polygon must contain at least one valid linestring"

        errors = []
        for val in value:
            error = self._validate_linestring(val, False)
            if not error and val[0] != val[-1]:
                error = "LineStrings must start and end at the same point"
            if error and error not in errors:
                errors.append(error)
        if errors:
            if top_level:
                return "Invalid Polygon:\n%s" % ", ".join(errors)
            else:
                return "%s" % ", ".join(errors)

    def _validate_linestring(self, value, top_level=True):
        """Validate a linestring."""
        if not isinstance(value, (list, tuple)):
            return "LineStrings must contain list of coordinate pairs"

        # Quick and dirty validator
        try:
            value[0][0]
        except (TypeError, IndexError):
            return "Invalid LineString must contain at least one valid point"

        errors = []
        for val in value:
            error = self._validate_point(val)
            if error and error not in errors:
                errors.append(error)
        if errors:
            if top_level:
                return "Invalid LineString:\n%s" % ", ".join(errors)
            else:
                return "%s" % ", ".join(errors)

    def _validate_point(self, value):
        """Validate each set of coords"""
        if not isinstance(value, (list, tuple)):
            return "Points must be a list of coordinate pairs"
        elif not len(value) == 2:
            return "Value (%s) must be a two-dimensional point" % repr(value)
        elif not isinstance(value[0], (float, int)) or not isinstance(
            value[1], (float, int)
        ):
            return "Both values (%s) in point must be float or int" % repr(value)

    def _validate_multipoint(self, value):
        if not isinstance(value, (list, tuple)):
            return "MultiPoint must be a list of Point"

        # Quick and dirty validator
        try:
            value[0][0]
        except (TypeError, IndexError):
            return "Invalid MultiPoint must contain at least one valid point"

        errors = []
        for point in value:
            error = self._validate_point(point)
            if error and error not in errors:
                errors.append(error)

        if errors:
            return "%s" % ", ".join(errors)

    def _validate_multilinestring(self, value, top_level=True):
        if not isinstance(value, (list, tuple)):
            return "MultiLineString must be a list of LineString"

        # Quick and dirty validator
        try:
            value[0][0][0]
        except (TypeError, IndexError):
            return "Invalid MultiLineString must contain at least one valid linestring"

        errors = []
        for linestring in value:
            error = self._validate_linestring(linestring, False)
            if error and error not in errors:
                errors.append(error)

        if errors:
            if top_level:
                return "Invalid MultiLineString:\n%s" % ", ".join(errors)
            else:
                return "%s" % ", ".join(errors)

    def _validate_multipolygon(self, value):
        if not isinstance(value, (list, tuple)):
            return "MultiPolygon must be a list of Polygon"

        # Quick and dirty validator
        try:
            value[0][0][0][0]
        except (TypeError, IndexError):
            return "Invalid MultiPolygon must contain at least one valid Polygon"

        errors = []
        for polygon in value:
            error = self._validate_polygon(polygon, False)
            if error and error not in errors:
                errors.append(error)

        if errors:
            return "Invalid MultiPolygon:\n%s" % ", ".join(errors)

    def to_mongo(self, value):
        if isinstance(value, dict):
            return value
        return SON([("type", self._type), ("coordinates", value)])

bagerard / mongoengine / 16950450531

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous