Method _get_capped_collection

mongoengine/document.py:239–274

Create a new or get an existing capped PyMongo collection.

(cls)

    @classmethod
    def _get_capped_collection(cls):
        """Create a new or get an existing capped PyMongo collection."""
        db = cls._get_db()
        collection_name = cls._get_collection_name()

        # Get max document limit and max byte size from meta.
        max_size = cls._meta.get("max_size") or 10 * 2**20  # 10MB default
        max_documents = cls._meta.get("max_documents")

        # MongoDB will automatically raise the size to make it a multiple of
        # 256 bytes. We raise it here ourselves to be able to reliably compare
        # the options below.
        if max_size % 256:
            max_size = (max_size // 256 + 1) * 256

        # If the collection already exists and has different options
        # (i.e. isn't capped or has different max/size), raise an error.
        if collection_name in list_collection_names(
            db, include_system_collections=True
        ):
            collection = db[collection_name]
            options = collection.options()
            if options.get("max") != max_documents or options.get("size") != max_size:
                raise InvalidCollectionError(
                    'Cannot create collection "{}" as a capped '
                    "collection as it already exists".format(cls._collection)
                )

            return collection

        # Create a new capped collection.
        opts = {"capped": True, "size": max_size}
        if max_documents:
            opts["max"] = max_documents

        return db.create_collection(collection_name, **opts)

    @classmethod
    def _get_timeseries_collection(cls):