Method _get_timeseries_collection

mongoengine/document.py:277–295

Create a new or get an existing timeseries PyMongo collection.

(cls)

    @classmethod
    def _get_timeseries_collection(cls):
        """Create a new or get an existing timeseries PyMongo collection."""
        db = cls._get_db()
        collection_name = cls._get_collection_name()
        timeseries_opts = cls._meta.get("timeseries")

        if collection_name in list_collection_names(
            db, include_system_collections=True
        ):
            collection = db[collection_name]
            collection.options()
            return collection

        opts = {"expireAfterSeconds": timeseries_opts.pop("expireAfterSeconds", None)}
        return db.create_collection(
            name=collection_name,
            timeseries=timeseries_opts,
            **opts,
        )

    def to_mongo(self, *args, **kwargs):
        data = super().to_mongo(*args, **kwargs)