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

94.42

/mongoengine/base/datastructures.py

import weakref

from bson import DBRef

from mongoengine.common import _import_class
from mongoengine.errors import DoesNotExist, MultipleObjectsReturned

__all__ = (
    "BaseDict",
    "StrictDict",
    "BaseList",
    "EmbeddedDocumentList",
    "LazyReference",
)


def mark_as_changed_wrapper(parent_method):
    """Decorator that ensures _mark_as_changed method gets called."""

    def wrapper(self, *args, **kwargs):
        # Can't use super() in the decorator.
        result = parent_method(self, *args, **kwargs)
        self._mark_as_changed()
        return result

    return wrapper


def mark_key_as_changed_wrapper(parent_method):
    """Decorator that ensures _mark_as_changed method gets called with the key argument"""

    def wrapper(self, key, *args, **kwargs):
        # Can't use super() in the decorator.
        if not args or key not in self or self[key] != args[0]:
            self._mark_as_changed(key)
        return parent_method(self, key, *args, **kwargs)

    return wrapper


class BaseDict(dict):
    """A special dict so we can watch any changes."""

    _dereferenced = False
    _instance = None
    _name = None

    def __init__(self, dict_items, instance, name):
        BaseDocument = _import_class("BaseDocument")

        if isinstance(instance, BaseDocument):
            self._instance = weakref.proxy(instance)
        self._name = name
        super().__init__(dict_items)

    def get(self, key, default=None):
        # get does not use __getitem__ by default so we must override it as well
        try:
            return self.__getitem__(key)
        except KeyError:
            return default

    def __getitem__(self, key):
        value = super().__getitem__(key)

        EmbeddedDocument = _import_class("EmbeddedDocument")
        if isinstance(value, EmbeddedDocument) and value._instance is None:
            value._instance = self._instance
        elif isinstance(value, dict) and not isinstance(value, BaseDict):
            value = BaseDict(value, None, f"{self._name}.{key}")
            super().__setitem__(key, value)
            value._instance = self._instance
        elif isinstance(value, list) and not isinstance(value, BaseList):
            value = BaseList(value, None, f"{self._name}.{key}")
            super().__setitem__(key, value)
            value._instance = self._instance
        return value

    def __getstate__(self):
        self.instance = None
        self._dereferenced = False
        return self

    def __setstate__(self, state):
        self = state
        return self

    __setitem__ = mark_key_as_changed_wrapper(dict.__setitem__)
    __delattr__ = mark_key_as_changed_wrapper(dict.__delattr__)
    __delitem__ = mark_key_as_changed_wrapper(dict.__delitem__)
    pop = mark_as_changed_wrapper(dict.pop)
    clear = mark_as_changed_wrapper(dict.clear)
    update = mark_as_changed_wrapper(dict.update)
    popitem = mark_as_changed_wrapper(dict.popitem)
    setdefault = mark_as_changed_wrapper(dict.setdefault)

    def _mark_as_changed(self, key=None):
        if hasattr(self._instance, "_mark_as_changed"):
            if key:
                self._instance._mark_as_changed(f"{self._name}.{key}")
            else:
                self._instance._mark_as_changed(self._name)


class BaseList(list):
    """A special list so we can watch any changes."""

    _dereferenced = False
    _instance = None
    _name = None

    def __init__(self, list_items, instance, name):
        BaseDocument = _import_class("BaseDocument")

        if isinstance(instance, BaseDocument):
            if isinstance(instance, weakref.ProxyTypes):
                self._instance = instance
            else:
                self._instance = weakref.proxy(instance)

        self._name = name
        super().__init__(list_items)

    def __getitem__(self, key):
        # change index to positive value because MongoDB does not support negative one
        if isinstance(key, int) and key < 0:
            key = len(self) + key
        value = super().__getitem__(key)

        if isinstance(key, slice):
            # When receiving a slice operator, we don't convert the structure and bind
            # to parent's instance. This is buggy for now but would require more work to be handled properly
            return value

        EmbeddedDocument = _import_class("EmbeddedDocument")
        if isinstance(value, EmbeddedDocument) and value._instance is None:
            value._instance = self._instance
        elif isinstance(value, dict) and not isinstance(value, BaseDict):
            # Replace dict by BaseDict
            value = BaseDict(value, None, f"{self._name}.{key}")
            super().__setitem__(key, value)
            value._instance = self._instance
        elif isinstance(value, list) and not isinstance(value, BaseList):
            # Replace list by BaseList
            value = BaseList(value, None, f"{self._name}.{key}")
            super().__setitem__(key, value)
            value._instance = self._instance
        return value

    def __iter__(self):
        yield from super().__iter__()

    def __getstate__(self):
        self.instance = None
        self._dereferenced = False
        return self

    def __setstate__(self, state):
        self = state
        return self

    def __setitem__(self, key, value):
        changed_key = key
        if isinstance(key, slice):
            # In case of slice, we don't bother to identify the exact elements being updated
            # instead, we simply marks the whole list as changed
            changed_key = None

        result = super().__setitem__(key, value)
        self._mark_as_changed(changed_key)
        return result

    append = mark_as_changed_wrapper(list.append)
    extend = mark_as_changed_wrapper(list.extend)
    insert = mark_as_changed_wrapper(list.insert)
    pop = mark_as_changed_wrapper(list.pop)
    remove = mark_as_changed_wrapper(list.remove)
    reverse = mark_as_changed_wrapper(list.reverse)
    sort = mark_as_changed_wrapper(list.sort)
    clear = mark_as_changed_wrapper(list.clear)
    __delitem__ = mark_as_changed_wrapper(list.__delitem__)
    __iadd__ = mark_as_changed_wrapper(list.__iadd__)
    __imul__ = mark_as_changed_wrapper(list.__imul__)

    def _mark_as_changed(self, key=None):
        if hasattr(self._instance, "_mark_as_changed"):
            if key is not None:
                self._instance._mark_as_changed(f"{self._name}.{key % len(self)}")
            else:
                self._instance._mark_as_changed(self._name)


class EmbeddedDocumentList(BaseList):
    @classmethod
    def __match_all(cls, embedded_doc, kwargs):
        """Return True if a given embedded doc matches all the filter
        kwargs. If it doesn't return False.
        """
        for key, expected_value in kwargs.items():
            doc_val = getattr(embedded_doc, key)
            if doc_val != expected_value and str(doc_val) != expected_value:
                return False
        return True

    @classmethod
    def __only_matches(cls, embedded_docs, kwargs):
        """Return embedded docs that match the filter kwargs."""
        if not kwargs:
            return embedded_docs
        return [doc for doc in embedded_docs if cls.__match_all(doc, kwargs)]

    def filter(self, **kwargs):
        """
        Filters the list by only including embedded documents with the
        given keyword arguments.

        This method only supports simple comparison (e.g. .filter(name='John Doe'))
        and does not support operators like __gte, __lte, __icontains like queryset.filter does

        :param kwargs: The keyword arguments corresponding to the fields to
         filter on. *Multiple arguments are treated as if they are ANDed
         together.*
        :return: A new ``EmbeddedDocumentList`` containing the matching
         embedded documents.

        Raises ``AttributeError`` if a given keyword is not a valid field for
        the embedded document class.
        """
        values = self.__only_matches(self, kwargs)
        return EmbeddedDocumentList(values, self._instance, self._name)

    def exclude(self, **kwargs):
        """
        Filters the list by excluding embedded documents with the given
        keyword arguments.

        :param kwargs: The keyword arguments corresponding to the fields to
         exclude on. *Multiple arguments are treated as if they are ANDed
         together.*
        :return: A new ``EmbeddedDocumentList`` containing the non-matching
         embedded documents.

        Raises ``AttributeError`` if a given keyword is not a valid field for
        the embedded document class.
        """
        exclude = self.__only_matches(self, kwargs)
        values = [item for item in self if item not in exclude]
        return EmbeddedDocumentList(values, self._instance, self._name)

    def count(self):
        """
        The number of embedded documents in the list.

        :return: The length of the list, equivalent to the result of ``len()``.
        """
        return len(self)

    def get(self, **kwargs):
        """
        Retrieves an embedded document determined by the given keyword
        arguments.

        :param kwargs: The keyword arguments corresponding to the fields to
         search on. *Multiple arguments are treated as if they are ANDed
         together.*
        :return: The embedded document matched by the given keyword arguments.

        Raises ``DoesNotExist`` if the arguments used to query an embedded
        document returns no results. ``MultipleObjectsReturned`` if more
        than one result is returned.
        """
        values = self.__only_matches(self, kwargs)
        if len(values) == 0:
            raise DoesNotExist("%s matching query does not exist." % self._name)
        elif len(values) > 1:
            raise MultipleObjectsReturned(
                "%d items returned, instead of 1" % len(values)
            )

        return values[0]

    def first(self):
        """Return the first embedded document in the list, or ``None``
        if empty.
        """
        if len(self) > 0:
            return self[0]

    def create(self, **values):
        """
        Creates a new instance of the EmbeddedDocument and appends it to this EmbeddedDocumentList.

        .. note::
            the instance of the EmbeddedDocument is not automatically saved to the database.
            You still need to call .save() on the parent Document.

        :param values: A dictionary of values for the embedded document.
        :return: The new embedded document instance.
        """
        name = self._name
        EmbeddedClass = self._instance._fields[name].field.document_type_obj
        self._instance[self._name].append(EmbeddedClass(**values))

        return self._instance[self._name][-1]

    def save(self, *args, **kwargs):
        """
        Saves the ancestor document.

        :param args: Arguments passed up to the ancestor Document's save
         method.
        :param kwargs: Keyword arguments passed up to the ancestor Document's
         save method.
        """
        self._instance.save(*args, **kwargs)

    def delete(self):
        """
        Deletes the embedded documents from the database.

        .. note::
            The embedded document changes are not automatically saved
            to the database after calling this method.

        :return: The number of entries deleted.
        """
        values = list(self)
        for item in values:
            self._instance[self._name].remove(item)

        return len(values)

    def update(self, **update):
        """
        Updates the embedded documents with the given replacement values. This
        function does not support mongoDB update operators such as ``inc__``.

        .. note::
            The embedded document changes are not automatically saved
            to the database after calling this method.

        :param update: A dictionary of update values to apply to each
         embedded document.
        :return: The number of entries updated.
        """
        if len(update) == 0:
            return 0
        values = list(self)
        for item in values:
            for k, v in update.items():
                setattr(item, k, v)

        return len(values)


class StrictDict:
    __slots__ = ()
    _special_fields = {"get", "pop", "iteritems", "items", "keys", "create"}
    _classes = {}

    def __init__(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)

    def __getitem__(self, key):
        key = "_reserved_" + key if key in self._special_fields else key
        try:
            return getattr(self, key)
        except AttributeError:
            raise KeyError(key)

    def __setitem__(self, key, value):
        key = "_reserved_" + key if key in self._special_fields else key
        return setattr(self, key, value)

    def __contains__(self, key):
        return hasattr(self, key)

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def pop(self, key, default=None):
        v = self.get(key, default)
        try:
            delattr(self, key)
        except AttributeError:
            pass
        return v

    def iteritems(self):
        for key in self:
            yield key, self[key]

    def items(self):
        return [(k, self[k]) for k in iter(self)]

    def iterkeys(self):
        return iter(self)

    def keys(self):
        return list(iter(self))

    def __iter__(self):
        return (key for key in self.__slots__ if hasattr(self, key))

    def __len__(self):
        return len(list(self.items()))

    def __eq__(self, other):
        return list(self.items()) == list(other.items())

    def __ne__(self, other):
        return not (self == other)

    @classmethod
    def create(cls, allowed_keys):
        allowed_keys_tuple = tuple(
            ("_reserved_" + k if k in cls._special_fields else k) for k in allowed_keys
        )
        allowed_keys = frozenset(allowed_keys_tuple)
        if allowed_keys not in cls._classes:

            class SpecificStrictDict(cls):
                __slots__ = allowed_keys_tuple

                def __repr__(self):
                    return "{%s}" % ", ".join(
                        f'"{k!s}": {v!r}' for k, v in self.items()
                    )

            cls._classes[allowed_keys] = SpecificStrictDict
        return cls._classes[allowed_keys]


class LazyReference(DBRef):
    __slots__ = ("_cached_doc", "passthrough", "document_type")

    def fetch(self, force=False):
        if not self._cached_doc or force:
            self._cached_doc = self.document_type.objects.get(pk=self.pk)
            if not self._cached_doc:
                raise DoesNotExist("Trying to dereference unknown document %s" % (self))
        return self._cached_doc

    @property
    def pk(self):
        return self.id

    def __init__(self, document_type, pk, cached_doc=None, passthrough=False):
        self.document_type = document_type
        self._cached_doc = cached_doc
        self.passthrough = passthrough
        super().__init__(self.document_type._get_collection_name(), pk)

    def __getitem__(self, name):
        if not self.passthrough:
            raise KeyError()
        document = self.fetch()
        return document[name]

    def __getattr__(self, name):
        if not object.__getattribute__(self, "passthrough"):
            raise AttributeError()
        document = self.fetch()
        try:
            return document[name]
        except KeyError:
            raise AttributeError()

    def __repr__(self):
        return f"<LazyReference({self.document_type}, {self.pk!r})>"

1	import weakref	2✔
2
3	from bson import DBRef	2✔
4
5	from mongoengine.common import _import_class	2✔
6	from mongoengine.errors import DoesNotExist, MultipleObjectsReturned	2✔
7
8	__all__ = (	2✔
9	"BaseDict",
10	"StrictDict",
11	"BaseList",
12	"EmbeddedDocumentList",
13	"LazyReference",
14	)
15
16
17	def mark_as_changed_wrapper(parent_method):	2✔
18	"""Decorator that ensures _mark_as_changed method gets called."""
19
20	def wrapper(self, args, *kwargs):	2✔
21	# Can't use super() in the decorator.
22	result = parent_method(self, args, *kwargs)	2✔
23	self._mark_as_changed()	2✔
24	return result	2✔
25
26	return wrapper	2✔
27
28
29	def mark_key_as_changed_wrapper(parent_method):	2✔
30	"""Decorator that ensures _mark_as_changed method gets called with the key argument"""
31
32	def wrapper(self, key, args, *kwargs):	2✔
33	# Can't use super() in the decorator.
34	if not args or key not in self or self[key] != args[0]:	2✔
35	self._mark_as_changed(key)	2✔
36	return parent_method(self, key, args, *kwargs)	2✔
37
38	return wrapper	2✔
39
40
41	class BaseDict(dict):	2✔
42	"""A special dict so we can watch any changes."""
43
44	_dereferenced = False	2✔
45	_instance = None	2✔
46	_name = None	2✔
47
48	def __init__(self, dict_items, instance, name):	2✔
49	BaseDocument = _import_class("BaseDocument")	2✔
50
51	if isinstance(instance, BaseDocument):	2✔
52	self._instance = weakref.proxy(instance)	2✔
53	self._name = name	2✔
54	super().__init__(dict_items)	2✔
55
56	def get(self, key, default=None):	2✔
57	# get does not use __getitem__ by default so we must override it as well
58	try:	2✔
59	return self.__getitem__(key)	2✔
60	except KeyError:	2✔
61	return default	2✔
62
63	def __getitem__(self, key):	2✔
64	value = super().__getitem__(key)	2✔
65
66	EmbeddedDocument = _import_class("EmbeddedDocument")	2✔
67	if isinstance(value, EmbeddedDocument) and value._instance is None:	2✔
68	value._instance = self._instance	2✔
69	elif isinstance(value, dict) and not isinstance(value, BaseDict):	2✔
70	value = BaseDict(value, None, f"{self._name}.{key}")	2✔
71	super().__setitem__(key, value)	2✔
72	value._instance = self._instance	2✔
73	elif isinstance(value, list) and not isinstance(value, BaseList):	2✔
74	value = BaseList(value, None, f"{self._name}.{key}")	2✔
75	super().__setitem__(key, value)	2✔
76	value._instance = self._instance	2✔
77	return value	2✔
78
79	def __getstate__(self):	2✔
80	self.instance = None	2✔
81	self._dereferenced = False	2✔
82	return self	2✔
83
84	def __setstate__(self, state):	2✔
85	self = state	2✔
86	return self	2✔
87
88	__setitem__ = mark_key_as_changed_wrapper(dict.__setitem__)	2✔
89	__delattr__ = mark_key_as_changed_wrapper(dict.__delattr__)	2✔
90	__delitem__ = mark_key_as_changed_wrapper(dict.__delitem__)	2✔
91	pop = mark_as_changed_wrapper(dict.pop)	2✔
92	clear = mark_as_changed_wrapper(dict.clear)	2✔
93	update = mark_as_changed_wrapper(dict.update)	2✔
94	popitem = mark_as_changed_wrapper(dict.popitem)	2✔
95	setdefault = mark_as_changed_wrapper(dict.setdefault)	2✔
96
97	def _mark_as_changed(self, key=None):	2✔
98	if hasattr(self._instance, "_mark_as_changed"):	2✔
99	if key:	2✔
100	self._instance._mark_as_changed(f"{self._name}.{key}")	2✔
101	else:
102	self._instance._mark_as_changed(self._name)	2✔
103
104
105	class BaseList(list):	2✔
106	"""A special list so we can watch any changes."""
107
108	_dereferenced = False	2✔
109	_instance = None	2✔
110	_name = None	2✔
111
112	def __init__(self, list_items, instance, name):	2✔
113	BaseDocument = _import_class("BaseDocument")	2✔
114
115	if isinstance(instance, BaseDocument):	2✔
116	if isinstance(instance, weakref.ProxyTypes):	2✔
117	self._instance = instance	2✔
118	else:
119	self._instance = weakref.proxy(instance)	2✔
120
121	self._name = name	2✔
122	super().__init__(list_items)	2✔
123
124	def __getitem__(self, key):	2✔
125	# change index to positive value because MongoDB does not support negative one
126	if isinstance(key, int) and key < 0:	2✔
127	key = len(self) + key	2✔
128	value = super().__getitem__(key)	2✔
129
130	if isinstance(key, slice):	2✔
131	# When receiving a slice operator, we don't convert the structure and bind
132	# to parent's instance. This is buggy for now but would require more work to be handled properly
133	return value	2✔
134
135	EmbeddedDocument = _import_class("EmbeddedDocument")	2✔
136	if isinstance(value, EmbeddedDocument) and value._instance is None:	2✔
137	value._instance = self._instance	2✔
138	elif isinstance(value, dict) and not isinstance(value, BaseDict):	2✔
139	# Replace dict by BaseDict
140	value = BaseDict(value, None, f"{self._name}.{key}")	2✔
141	super().__setitem__(key, value)	2✔
142	value._instance = self._instance	2✔
143	elif isinstance(value, list) and not isinstance(value, BaseList):	2✔
144	# Replace list by BaseList
145	value = BaseList(value, None, f"{self._name}.{key}")	2✔
146	super().__setitem__(key, value)	2✔
147	value._instance = self._instance	2✔
148	return value	2✔
149
150	def __iter__(self):	2✔
151	yield from super().__iter__()	2✔
152
153	def __getstate__(self):	2✔
154	self.instance = None	×
155	self._dereferenced = False	×
156	return self	×
157
158	def __setstate__(self, state):	2✔
159	self = state	×
160	return self	×
161
162	def __setitem__(self, key, value):	2✔
163	changed_key = key	2✔
164	if isinstance(key, slice):	2✔
165	# In case of slice, we don't bother to identify the exact elements being updated
166	# instead, we simply marks the whole list as changed
167	changed_key = None	2✔
168
169	result = super().__setitem__(key, value)	2✔
170	self._mark_as_changed(changed_key)	2✔
171	return result	2✔
172
173	append = mark_as_changed_wrapper(list.append)	2✔
174	extend = mark_as_changed_wrapper(list.extend)	2✔
175	insert = mark_as_changed_wrapper(list.insert)	2✔
176	pop = mark_as_changed_wrapper(list.pop)	2✔
177	remove = mark_as_changed_wrapper(list.remove)	2✔
178	reverse = mark_as_changed_wrapper(list.reverse)	2✔
179	sort = mark_as_changed_wrapper(list.sort)	2✔
180	clear = mark_as_changed_wrapper(list.clear)	2✔
181	__delitem__ = mark_as_changed_wrapper(list.__delitem__)	2✔
182	__iadd__ = mark_as_changed_wrapper(list.__iadd__)	2✔
183	__imul__ = mark_as_changed_wrapper(list.__imul__)	2✔
184
185	def _mark_as_changed(self, key=None):	2✔
186	if hasattr(self._instance, "_mark_as_changed"):	2✔
187	if key is not None:	2✔
188	self._instance._mark_as_changed(f"{self._name}.{key % len(self)}")	2✔
189	else:
190	self._instance._mark_as_changed(self._name)	2✔
191
192
193	class EmbeddedDocumentList(BaseList):	2✔
194	@classmethod	2✔
195	def __match_all(cls, embedded_doc, kwargs):	2✔
196	"""Return True if a given embedded doc matches all the filter
197	kwargs. If it doesn't return False.
198	"""
199	for key, expected_value in kwargs.items():	2✔
200	doc_val = getattr(embedded_doc, key)	2✔
201	if doc_val != expected_value and str(doc_val) != expected_value:	2✔
202	return False	2✔
203	return True	2✔
204
205	@classmethod	2✔
206	def __only_matches(cls, embedded_docs, kwargs):	2✔
207	"""Return embedded docs that match the filter kwargs."""
208	if not kwargs:	2✔
209	return embedded_docs	2✔
210	return [doc for doc in embedded_docs if cls.__match_all(doc, kwargs)]	2✔
211
212	def filter(self, **kwargs):	2✔
213	"""
214	Filters the list by only including embedded documents with the
215	given keyword arguments.
216
217	This method only supports simple comparison (e.g. .filter(name='John Doe'))
218	and does not support operators like __gte, __lte, __icontains like queryset.filter does
219
220	:param kwargs: The keyword arguments corresponding to the fields to
221	filter on. *Multiple arguments are treated as if they are ANDed
222	together.*
223	:return: A new ``EmbeddedDocumentList`` containing the matching
224	embedded documents.
225
226	Raises ``AttributeError`` if a given keyword is not a valid field for
227	the embedded document class.
228	"""
229	values = self.__only_matches(self, kwargs)	2✔
230	return EmbeddedDocumentList(values, self._instance, self._name)	2✔
231
232	def exclude(self, **kwargs):	2✔
233	"""
234	Filters the list by excluding embedded documents with the given
235	keyword arguments.
236
237	:param kwargs: The keyword arguments corresponding to the fields to
238	exclude on. *Multiple arguments are treated as if they are ANDed
239	together.*
240	:return: A new ``EmbeddedDocumentList`` containing the non-matching
241	embedded documents.
242
243	Raises ``AttributeError`` if a given keyword is not a valid field for
244	the embedded document class.
245	"""
246	exclude = self.__only_matches(self, kwargs)	2✔
247	values = [item for item in self if item not in exclude]	2✔
248	return EmbeddedDocumentList(values, self._instance, self._name)	2✔
249
250	def count(self):	2✔
251	"""
252	The number of embedded documents in the list.
253
254	:return: The length of the list, equivalent to the result of ``len()``.
255	"""
256	return len(self)	2✔
257
258	def get(self, **kwargs):	2✔
259	"""
260	Retrieves an embedded document determined by the given keyword
261	arguments.
262
263	:param kwargs: The keyword arguments corresponding to the fields to
264	search on. *Multiple arguments are treated as if they are ANDed
265	together.*
266	:return: The embedded document matched by the given keyword arguments.
267
268	Raises ``DoesNotExist`` if the arguments used to query an embedded
269	document returns no results. ``MultipleObjectsReturned`` if more
270	than one result is returned.
271	"""
272	values = self.__only_matches(self, kwargs)	2✔
273	if len(values) == 0:	2✔
274	raise DoesNotExist("%s matching query does not exist." % self._name)	2✔
275	elif len(values) > 1:	2✔
276	raise MultipleObjectsReturned(	2✔
277	"%d items returned, instead of 1" % len(values)
278	)
279
280	return values[0]	2✔
281
282	def first(self):	2✔
283	"""Return the first embedded document in the list, or ``None``
284	if empty.
285	"""
286	if len(self) > 0:	2✔
287	return self[0]	2✔
288
289	def create(self, **values):	2✔
290	"""
291	Creates a new instance of the EmbeddedDocument and appends it to this EmbeddedDocumentList.
292
293	.. note::
294	the instance of the EmbeddedDocument is not automatically saved to the database.
295	You still need to call .save() on the parent Document.
296
297	:param values: A dictionary of values for the embedded document.
298	:return: The new embedded document instance.
299	"""
300	name = self._name	2✔
301	EmbeddedClass = self._instance._fields[name].field.document_type_obj	2✔
302	self._instance[self._name].append(EmbeddedClass(**values))	2✔
303
304	return self._instance[self._name][-1]	2✔
305
306	def save(self, args, *kwargs):	2✔
307	"""
308	Saves the ancestor document.
309
310	:param args: Arguments passed up to the ancestor Document's save
311	method.
312	:param kwargs: Keyword arguments passed up to the ancestor Document's
313	save method.
314	"""
315	self._instance.save(args, *kwargs)	2✔
316
317	def delete(self):	2✔
318	"""
319	Deletes the embedded documents from the database.
320
321	.. note::
322	The embedded document changes are not automatically saved
323	to the database after calling this method.
324
325	:return: The number of entries deleted.
326	"""
327	values = list(self)	2✔
328	for item in values:	2✔
329	self._instance[self._name].remove(item)	2✔
330
331	return len(values)	2✔
332
333	def update(self, **update):	2✔
334	"""
335	Updates the embedded documents with the given replacement values. This
336	function does not support mongoDB update operators such as ``inc__``.
337
338	.. note::
339	The embedded document changes are not automatically saved
340	to the database after calling this method.
341
342	:param update: A dictionary of update values to apply to each
343	embedded document.
344	:return: The number of entries updated.
345	"""
346	if len(update) == 0:	2✔
347	return 0	2✔
348	values = list(self)	2✔
349	for item in values:	2✔
350	for k, v in update.items():	2✔
351	setattr(item, k, v)	2✔
352
353	return len(values)	2✔
354
355
356	class StrictDict:	2✔
357	__slots__ = ()	2✔
358	_special_fields = {"get", "pop", "iteritems", "items", "keys", "create"}	2✔
359	_classes = {}	2✔
360
361	def __init__(self, **kwargs):	2✔
362	for k, v in kwargs.items():	2✔
363	setattr(self, k, v)	2✔
364
365	def __getitem__(self, key):	2✔
366	key = "_reserved_" + key if key in self._special_fields else key	2✔
367	try:	2✔
368	return getattr(self, key)	2✔
369	except AttributeError:	2✔
370	raise KeyError(key)	2✔
371
372	def __setitem__(self, key, value):	2✔
373	key = "_reserved_" + key if key in self._special_fields else key	×
374	return setattr(self, key, value)	×
375
376	def __contains__(self, key):	2✔
377	return hasattr(self, key)	2✔
378
379	def get(self, key, default=None):	2✔
380	try:	2✔
381	return self[key]	2✔
382	except KeyError:	2✔
383	return default	2✔
384
385	def pop(self, key, default=None):	2✔
386	v = self.get(key, default)	2✔
387	try:	2✔
388	delattr(self, key)	2✔
389	except AttributeError:	×
390	pass	×
391	return v	2✔
392
393	def iteritems(self):	2✔
394	for key in self:	×
395	yield key, self[key]	×
396
397	def items(self):	2✔
398	return [(k, self[k]) for k in iter(self)]	2✔
399
400	def iterkeys(self):	2✔
401	return iter(self)	×
402
403	def keys(self):	2✔
404	return list(iter(self))	2✔
405
406	def __iter__(self):	2✔
407	return (key for key in self.__slots__ if hasattr(self, key))	2✔
408
409	def __len__(self):	2✔
410	return len(list(self.items()))	2✔
411
412	def __eq__(self, other):	2✔
413	return list(self.items()) == list(other.items())	2✔
414
415	def __ne__(self, other):	2✔
416	return not (self == other)	2✔
417
418	@classmethod	2✔
419	def create(cls, allowed_keys):	2✔
420	allowed_keys_tuple = tuple(	2✔
421	("_reserved_" + k if k in cls._special_fields else k) for k in allowed_keys
422	)
423	allowed_keys = frozenset(allowed_keys_tuple)	2✔
424	if allowed_keys not in cls._classes:	2✔
425
426	class SpecificStrictDict(cls):	2✔
427	__slots__ = allowed_keys_tuple	2✔
428
429	def __repr__(self):	2✔
430	return "{%s}" % ", ".join(	2✔
431	f'"{k!s}": {v!r}' for k, v in self.items()
432	)
433
434	cls._classes[allowed_keys] = SpecificStrictDict	2✔
435	return cls._classes[allowed_keys]	2✔
436
437
438	class LazyReference(DBRef):	2✔
439	__slots__ = ("_cached_doc", "passthrough", "document_type")	2✔
440
441	def fetch(self, force=False):	2✔
442	if not self._cached_doc or force:	2✔
443	self._cached_doc = self.document_type.objects.get(pk=self.pk)	2✔
444	if not self._cached_doc:	2✔
445	raise DoesNotExist("Trying to dereference unknown document %s" % (self))	×
446	return self._cached_doc	2✔
447
448	@property	2✔
449	def pk(self):	2✔
450	return self.id	2✔
451
452	def __init__(self, document_type, pk, cached_doc=None, passthrough=False):	2✔
453	self.document_type = document_type	2✔
454	self._cached_doc = cached_doc	2✔
455	self.passthrough = passthrough	2✔
456	super().__init__(self.document_type._get_collection_name(), pk)	2✔
457
458	def __getitem__(self, name):	2✔
459	if not self.passthrough:	2✔
460	raise KeyError()	2✔
461	document = self.fetch()	2✔
462	return document[name]	2✔
463
464	def __getattr__(self, name):	2✔
465	if not object.__getattribute__(self, "passthrough"):	2✔
466	raise AttributeError()	2✔
467	document = self.fetch()	2✔
468	try:	2✔
469	return document[name]	2✔
470	except KeyError:	×
471	raise AttributeError()	×
472
473	def __repr__(self):	2✔
474	return f"<LazyReference({self.document_type}, {self.pk!r})>"	2✔

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