15811502491

Committed 22 Jun 2025 10:36PM UTC coverage: 96.107%. First build

Build # 15811502491

Build Type

Pull #16

github

Committed by

web-flow

Commit Message

Merge a5cc0f810 into 085991d99

Pull Request Pull Request #16: refactor(humbledb): support for pymongo 4.x

Run Details

267 of 294 new or added lines in 10 files covered. (90.82%)

1185 of 1233 relevant lines covered (96.11%)

3.84 hits per line

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

93.79

/humbledb/array.py

import itertools

import humbledb
from humbledb import UNSET, Document, _version
from humbledb.errors import NoConnection


class Page(Document):
    """Document class used by :class:`Array`."""

    size = "s"  # Number of entries in this page
    """ Number of entries currently in this page. """
    entries = "e"  # Array of entries
    """ Array of entries. """
    _opts = {"safe": True} if _version._lt("3.0.0") else {}


class ArrayMeta(type):
    """
    Metaclass for Arrays. This ensures that we have all the needed
    configuration options, as well as creating the :class:`Page` subclass that
    is specific to each Array subclass.

    """

    def __new__(mcs, name, bases, cls_dict):
        # Skip the Array base class
        if (
            name == "Array"
            and not len(bases)
            and mcs is ArrayMeta
            and cls_dict["__qualname__"] == "Array"
        ):
            return type.__new__(mcs, name, bases, cls_dict)
        # The dictionary for subclassing the Page document
        page_dict = {}
        # Check for required class members
        for member in "config_database", "config_collection":
            if member not in cls_dict:
                raise TypeError("{!r} missing required {!r}".format(name, member))
            # Move the config to the page
            page_dict[member] = cls_dict.pop(member)
        # Create our page subclass and assign to cls._page
        cls_dict["_page"] = type(name + "Page", (Page,), page_dict)
        # Return our new Array
        return type.__new__(mcs, name, bases, cls_dict)

    # Shortcut methods
    @property
    def size(cls):
        return cls._page.size

    @property
    def entries(cls):
        return cls._page.entries

    @property
    def find(cls):
        return cls._page.find

    @property
    def update(cls):
        return cls._page.update

    @property
    def remove(cls):  # This needs a try/except for tests
        try:
            return cls._page.remove
        except NoConnection:
            pass  # Collection not available yet


class Array(metaclass=ArrayMeta):
    """
    HumbleDB Array object. This helps manage paginated array documents in
    MongoDB. This class is designed to be inherited from, and not instantiated
    directly.

    If you know the `page_count` for this array ahead of time, passing it in
    to the constructor will save an extra query on the first append for a given
    instance.

    :param str _id: Sets the array's shared id
    :param int page_count: Total number of pages that already exist (optional)

    """

    config_max_size = 100
    """ Soft limit on the maximum number of entries per page. """

    config_page_marker = "#"
    """ Combined with the array_id and page number to create the page _id. """

    config_padding = 0
    """ Number of bytes to pad new page creation with. """

    def __init__(self, _id, page_count=UNSET):
        self._array_id = _id
        self.page_count = page_count

    def page_id(self, page_number=None):
        """
        Return the document ID for `page_number`. If page number is not
        specified the :attr:`Array.page_count` is used.

        :param int page_number: A page number (optional)

        """
        page_number = page_number or self.page_count or 0
        return "{}{:05d}".format(self._id, page_number)

    @property
    def _id(self):
        return "{}{}".format(self._array_id, self.config_page_marker)

    @property
    def _id_regex(self):
        _id = self._id.replace(".", "\.")
        return {"$regex": "^" + _id}

    def new_page(self, page_number):
        """
        Creates a new page document.

        :param int page_number: The page number to create

        """
        # Shortcut the page class
        Page = self._page
        # Create a new page instance
        page = Page()
        page._id = self.page_id(page_number)
        page.size = 0
        page.entries = []
        page["padding"] = "0" * self.config_padding
        # Insert the new page
        try:
            # We need to do this as safe, because otherwise it may not be
            # available to a subsequent call to append
            Page.insert(page, **Page._opts)
        except humbledb.errors.DuplicateKeyError:
            # A race condition already created this page, so we are done
            return
        # Remove the padding
        Page.update({"_id": page._id}, {"$unset": {"padding": 1}}, **Page._opts)

    def append(self, entry):
        """
        Append an entry to this array and return the page count.

        :param dict entry: New entry
        :returns: Total number of pages

        """
        # If we haven't set a page count, we query for it. This is generally a
        # very fast query.
        if self.page_count is UNSET:
            self.page_count = self.pages()
        # See if we have to create our initial page
        if self.page_count < 1:
            self.page_count = 1
            self.new_page(self.page_count)
        # Shortcut page class
        Page = self._page
        query = {"_id": self.page_id()}
        modify = {"$inc": {Page.size: 1}, "$push": {Page.entries: entry}}
        fields = {Page.size: 1}
        # Append our entry to our page and get the page's size
        page = Page.find_and_modify(query, modify, new=True, fields=fields)
        if not page:
            raise RuntimeError("Append failed: page does not exist.")
        # If we need to, we create the next page
        if page.size >= self.config_max_size:
            self.page_count += 1
            self.new_page(self.page_count)
        # Return the page count
        return self.page_count

    def remove(self, spec):
        """
        Remove first element matching `spec` from each page in this array.

        Due to how this is handled, all ``null`` values will be removed from
        the array.

        :param dict spec: Dictionary matching items to be removed
        :returns: ``True`` if an element was removed

        """
        Page = self._page
        # Since we can't reliably use dot-notation when the query is against an
        # embedded document, we need to use the $elemMatch operator instead
        if isinstance(spec, dict):
            query_spec = {"$elemMatch": spec}
        else:
            query_spec = spec
        # Update to set first instance matching ``spec`` on each page to
        # ``null`` (via $unset)
        query = {"_id": self._id_regex, Page.entries: query_spec}
        modify = {"$unset": {Page.entries + ".$": spec}, "$inc": {Page.size: -1}}
        result = Page.update(query, modify, multi=True)
        if not result or not result.get("updatedExisting", None):
            return
        # Update to remove all ``null`` entries from this array
        query = {"_id": self._id_regex, Page.entries: None}
        result = Page.update(query, {"$pull": {Page.entries: None}}, multi=True)
        # Check the result and return True if anything was modified
        if result and result.get("updatedExisting", None):
            return True

    def _all(self):
        """Return a cursor for iterating over all the pages."""
        Page = self._page
        return Page.find({"_id": self._id_regex}).sort("_id")

    def all(self):
        """Return all entries in this array."""
        cursor = self._all()
        return list(itertools.chain.from_iterable(p.entries for p in cursor))

    def clear(self):
        """Remove all documents in this array."""
        self._page.remove({self._page._id: self._id_regex})
        self.page_count = 0

    def length(self):
        """Return the total number of items in this array."""
        # This is implemented rather than __len__ because it incurs a query,
        # and we don't want to query transparently
        Page = self._page
        if _version._lt("3.0.0"):
            cursor = Page.find({"_id": self._id_regex}, fields={Page.size: 1, "_id": 0})
        else:
            cursor = Page.find({"_id": self._id_regex}, {Page.size: 1, "_id": 0})
        return sum(p.size for p in cursor)

    def pages(self):
        """Return the total number of pages in this array."""
        Page = self._page
        return Page.find({"_id": self._id_regex}).count()

    def __getitem__(self, index):
        """
        Return a page or pages for the given index or slice respectively.

        :param index: Integer index or ``slice()`` object

        """
        if not isinstance(index, (int, slice)):
            raise TypeError("Array indices must be integers, not %s" % type(index))
        Page = self._page  # Shorthand the Page class
        # If we have an integer index, it's a simple query for the page number
        if isinstance(index, int):
            if index < 0:
                raise IndexError("Array indices must be positive")
            # Page numbers are not zero indexed
            index += 1
            page = Page.find_one({"_id": self.page_id(index)})
            if not page:
                raise IndexError("Array index out of range")
            return page.entries
        # If we have a slice, we attempt to get the pages for [start, stop)
        if isinstance(index, slice):
            if index.step:
                raise TypeError("Arrays do not allow extended slices")
            if index.start and index.start < 0:
                raise IndexError("Array indices must be positive")
            if index.stop and index.stop < 0:
                raise IndexError("Array indices must be positive")
            # Page numbers are not zero indexed
            start = (index.start or 0) + 1
            stop = (index.stop or 2**32) + 1
            start = "{}{:05d}".format(self._id, start)
            stop = "{}{:05d}".format(self._id, stop)
            cursor = Page.find({"_id": {"$gte": start, "$lt": stop}})
            return list(itertools.chain.from_iterable(p.entries for p in cursor))
        # This comment will never be reached

1	import itertools	4✔
2
3	import humbledb	4✔
4	from humbledb import UNSET, Document, _version	4✔
5	from humbledb.errors import NoConnection	4✔
6
7
8	class Page(Document):	4✔
9	"""Document class used by :class:`Array`."""
10
11	size = "s" # Number of entries in this page	4✔
12	""" Number of entries currently in this page. """	4✔
13	entries = "e" # Array of entries	4✔
14	""" Array of entries. """	4✔
15	_opts = {"safe": True} if _version._lt("3.0.0") else {}	4✔
16
17
18	class ArrayMeta(type):	4✔
19	"""
20	Metaclass for Arrays. This ensures that we have all the needed
21	configuration options, as well as creating the :class:`Page` subclass that
22	is specific to each Array subclass.
23
24	"""
25
26	def __new__(mcs, name, bases, cls_dict):	4✔
27	# Skip the Array base class
28	if (	4✔
29	name == "Array"
30	and not len(bases)
31	and mcs is ArrayMeta
32	and cls_dict["__qualname__"] == "Array"
33	):
34	return type.__new__(mcs, name, bases, cls_dict)	4✔
35	# The dictionary for subclassing the Page document
36	page_dict = {}	4✔
37	# Check for required class members
38	for member in "config_database", "config_collection":	4✔
39	if member not in cls_dict:	4✔
40	raise TypeError("{!r} missing required {!r}".format(name, member))	4✔
41	# Move the config to the page
42	page_dict[member] = cls_dict.pop(member)	4✔
43	# Create our page subclass and assign to cls._page
44	cls_dict["_page"] = type(name + "Page", (Page,), page_dict)	4✔
45	# Return our new Array
46	return type.__new__(mcs, name, bases, cls_dict)	4✔
47
48	# Shortcut methods
49	@property	4✔
50	def size(cls):	4✔
51	return cls._page.size	4✔
52
53	@property	4✔
54	def entries(cls):	4✔
55	return cls._page.entries	4✔
56
57	@property	4✔
58	def find(cls):	4✔
59	return cls._page.find	4✔
60
61	@property	4✔
62	def update(cls):	4✔
NEW 63	return cls._page.update	×
64
65	@property	4✔
66	def remove(cls): # This needs a try/except for tests	4✔
NEW 67	try:	×
NEW 68	return cls._page.remove	×
NEW 69	except NoConnection:	×
NEW 70	pass # Collection not available yet	×
71
72
73	class Array(metaclass=ArrayMeta):	4✔
74	"""
75	HumbleDB Array object. This helps manage paginated array documents in
76	MongoDB. This class is designed to be inherited from, and not instantiated
77	directly.
78
79	If you know the `page_count` for this array ahead of time, passing it in
80	to the constructor will save an extra query on the first append for a given
81	instance.
82
83	:param str _id: Sets the array's shared id
84	:param int page_count: Total number of pages that already exist (optional)
85
86	"""
87
88	config_max_size = 100	4✔
89	""" Soft limit on the maximum number of entries per page. """	4✔
90
91	config_page_marker = "#"	4✔
92	""" Combined with the array_id and page number to create the page _id. """	4✔
93
94	config_padding = 0	4✔
95	""" Number of bytes to pad new page creation with. """	4✔
96
97	def __init__(self, _id, page_count=UNSET):	4✔
98	self._array_id = _id	4✔
99	self.page_count = page_count	4✔
100
101	def page_id(self, page_number=None):	4✔
102	"""
103	Return the document ID for `page_number`. If page number is not
104	specified the :attr:`Array.page_count` is used.
105
106	:param int page_number: A page number (optional)
107
108	"""
109	page_number = page_number or self.page_count or 0	4✔
110	return "{}{:05d}".format(self._id, page_number)	4✔
111
112	@property	4✔
113	def _id(self):	4✔
114	return "{}{}".format(self._array_id, self.config_page_marker)	4✔
115
116	@property	4✔
117	def _id_regex(self):	4✔
118	_id = self._id.replace(".", "\.")	4✔
119	return {"$regex": "^" + _id}	4✔
120
121	def new_page(self, page_number):	4✔
122	"""
123	Creates a new page document.
124
125	:param int page_number: The page number to create
126
127	"""
128	# Shortcut the page class
129	Page = self._page	4✔
130	# Create a new page instance
131	page = Page()	4✔
132	page._id = self.page_id(page_number)	4✔
133	page.size = 0	4✔
134	page.entries = []	4✔
135	page["padding"] = "0" * self.config_padding	4✔
136	# Insert the new page
137	try:	4✔
138	# We need to do this as safe, because otherwise it may not be
139	# available to a subsequent call to append
140	Page.insert(page, **Page._opts)	4✔
141	except humbledb.errors.DuplicateKeyError:	4✔
142	# A race condition already created this page, so we are done
143	return	4✔
144	# Remove the padding
145	Page.update({"_id": page._id}, {"$unset": {"padding": 1}}, **Page._opts)	4✔
146
147	def append(self, entry):	4✔
148	"""
149	Append an entry to this array and return the page count.
150
151	:param dict entry: New entry
152	:returns: Total number of pages
153
154	"""
155	# If we haven't set a page count, we query for it. This is generally a
156	# very fast query.
157	if self.page_count is UNSET:	4✔
158	self.page_count = self.pages()	4✔
159	# See if we have to create our initial page
160	if self.page_count < 1:	4✔
161	self.page_count = 1	4✔
162	self.new_page(self.page_count)	4✔
163	# Shortcut page class
164	Page = self._page	4✔
165	query = {"_id": self.page_id()}	4✔
166	modify = {"$inc": {Page.size: 1}, "$push": {Page.entries: entry}}	4✔
167	fields = {Page.size: 1}	4✔
168	# Append our entry to our page and get the page's size
169	page = Page.find_and_modify(query, modify, new=True, fields=fields)	4✔
170	if not page:	4✔
171	raise RuntimeError("Append failed: page does not exist.")	4✔
172	# If we need to, we create the next page
173	if page.size >= self.config_max_size:	4✔
174	self.page_count += 1	4✔
175	self.new_page(self.page_count)	4✔
176	# Return the page count
177	return self.page_count	4✔
178
179	def remove(self, spec):	4✔
180	"""
181	Remove first element matching `spec` from each page in this array.
182
183	Due to how this is handled, all ``null`` values will be removed from
184	the array.
185
186	:param dict spec: Dictionary matching items to be removed
187	:returns: ``True`` if an element was removed
188
189	"""
190	Page = self._page	4✔
191	# Since we can't reliably use dot-notation when the query is against an
192	# embedded document, we need to use the $elemMatch operator instead
193	if isinstance(spec, dict):	4✔
194	query_spec = {"$elemMatch": spec}	4✔
195	else:
196	query_spec = spec	4✔
197	# Update to set first instance matching ``spec`` on each page to
198	# ``null`` (via $unset)
199	query = {"_id": self._id_regex, Page.entries: query_spec}	4✔
200	modify = {"$unset": {Page.entries + ".$": spec}, "$inc": {Page.size: -1}}	4✔
201	result = Page.update(query, modify, multi=True)	4✔
202	if not result or not result.get("updatedExisting", None):	4✔
203	return	4✔
204	# Update to remove all ``null`` entries from this array
205	query = {"_id": self._id_regex, Page.entries: None}	4✔
206	result = Page.update(query, {"$pull": {Page.entries: None}}, multi=True)	4✔
207	# Check the result and return True if anything was modified
208	if result and result.get("updatedExisting", None):	4✔
209	return True	4✔
210
211	def _all(self):	4✔
212	"""Return a cursor for iterating over all the pages."""
213	Page = self._page	4✔
214	return Page.find({"_id": self._id_regex}).sort("_id")	4✔
215
216	def all(self):	4✔
217	"""Return all entries in this array."""
218	cursor = self._all()	4✔
219	return list(itertools.chain.from_iterable(p.entries for p in cursor))	4✔
220
221	def clear(self):	4✔
222	"""Remove all documents in this array."""
223	self._page.remove({self._page._id: self._id_regex})	4✔
224	self.page_count = 0	4✔
225
226	def length(self):	4✔
227	"""Return the total number of items in this array."""
228	# This is implemented rather than __len__ because it incurs a query,
229	# and we don't want to query transparently
230	Page = self._page	4✔
231	if _version._lt("3.0.0"):	4✔
NEW 232	cursor = Page.find({"_id": self._id_regex}, fields={Page.size: 1, "_id": 0})	×
233	else:
234	cursor = Page.find({"_id": self._id_regex}, {Page.size: 1, "_id": 0})	4✔
235	return sum(p.size for p in cursor)	4✔
236
237	def pages(self):	4✔
238	"""Return the total number of pages in this array."""
239	Page = self._page	4✔
240	return Page.find({"_id": self._id_regex}).count()	4✔
241
242	def __getitem__(self, index):	4✔
243	"""
244	Return a page or pages for the given index or slice respectively.
245
246	:param index: Integer index or ``slice()`` object
247
248	"""
249	if not isinstance(index, (int, slice)):	4✔
250	raise TypeError("Array indices must be integers, not %s" % type(index))	4✔
251	Page = self._page # Shorthand the Page class	4✔
252	# If we have an integer index, it's a simple query for the page number
253	if isinstance(index, int):	4✔
254	if index < 0:	4✔
255	raise IndexError("Array indices must be positive")	×
256	# Page numbers are not zero indexed
257	index += 1	4✔
258	page = Page.find_one({"_id": self.page_id(index)})	4✔
259	if not page:	4✔
260	raise IndexError("Array index out of range")	4✔
261	return page.entries	4✔
262	# If we have a slice, we attempt to get the pages for [start, stop)
263	if isinstance(index, slice):	4✔
264	if index.step:	4✔
265	raise TypeError("Arrays do not allow extended slices")	4✔
266	if index.start and index.start < 0:	4✔
267	raise IndexError("Array indices must be positive")	×
268	if index.stop and index.stop < 0:	4✔
269	raise IndexError("Array indices must be positive")	×
270	# Page numbers are not zero indexed
271	start = (index.start or 0) + 1	4✔
272	stop = (index.stop or 2**32) + 1	4✔
273	start = "{}{:05d}".format(self._id, start)	4✔
274	stop = "{}{:05d}".format(self._id, stop)	4✔
275	cursor = Page.find({"_id": {"$gte": start, "$lt": stop}})	4✔
276	return list(itertools.chain.from_iterable(p.entries for p in cursor))	4✔
277	# This comment will never be reached

shakefu / humbledb / 15811502491

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