Python REST API and Mongo - Aggregation Pipeline/Stage classes

This has been my approach for "simplifying" MongoDB aggregation queries in a pythonic syntax – was my intent, at the very least:

from datetime import datetime, timedelta

class MatchStage(list):
    def _get_day_range(self):
        current_time = datetime.now()
        return (current_time - timedelta(days=1)), current_time
    def diferent(self, **kwargs):
        for field, value in kwargs.items():
            self.append({field: {"$ne": value}})
    def equals(self, **kwargs):
        for field, value in kwargs.items():
            self.append({field: {"$eq": value}})
    def set_interval(self, field, data):
        self.starttime, self.endtime = map(datetime.fromtimestamp, ( int(data["starttime"]), int(data["endtime"]) )) 
            if {"starttime", "endtime"} <= data.keys() else self._get_day_range()
        self.append({
            field: {
                "$gte": self.starttime,
                "$lt": self.endtime
            }
        })
    def set_devices(self, devices):
        self.devices = devices
        self.append({"device": {"$in": [device.id for device in devices]}})

class SortStage(dict):
    @staticmethod
    def ascending_order(*fields):
        return {"$sort": {field: 1 for field in fields} }
    @staticmethod
    def descending_order(*fields):
        return {"$sort": {field: -1 for field in fields} }

class ReRootStage(dict):
    @staticmethod
    def reset_root(*args):
        return {"$replaceRoot": {"newRoot": {"$mergeObjects": [{key: '$' + key for key in args}, "$_id"]}}}

class GroupStage(dict):
    def sum_metrics(self, **kwargs):
        self.update({k: {"$sum": v if isinstance(v, int) else {"$toLong": v if v.startswith('$') else '$' + v}} for k, v in kwargs.items()})
    def group_fields(self, *fields):
        self["_id"].update({field: '$' + field for field in fields})
    def _nest_field(self, key, field):
        self[key] = {"$push": field}
    def nest_field(self, key, field):
        if not field.startswith('$'): field = '$' + field
        self._nest_field(key, field)
    def nest_fields(self, *fields, key, **kfields):
        if not fields and not kfields: raise Exception("No field specified")
        self._nest_field(key, {field: '$' + field for field in fields} or {field: '$' + value for field, value in kfields.items()})
    
class BasePipeline(list):
    _match = MatchStage
    _group = GroupStage
    _sort = SortStage
    _reroot = ReRootStage
    def __init__(self, data, devices):
        self._set_match().set_interval("time", data)
        if "devid" in data:
            devices = devices.filter(id__in=data["devid"].split(','))
            assert devices.exists(), "Device not found"
        self.match.set_devices(devices)
    def _set_group(self):
        self.group = self._group()
        self.append({"$group": self.group})
        return self.group
    def _set_match(self):
        self.match = self._match()
        self.append({"$match": {"$and": self.match}})
        return self.match
    def set_simple_group(self, field):
        self._set_group().update({"_id": field if field.startswith('$') else '$' + field})
        self.extend([{"$addFields":{"id": "$_id"}}, {"$project": {"_id": 0}}])
    def set_composite_group(self, *fields, **kfields):
        if not fields and not kfields:
            raise Exception("No fields specified")
        self._set_group().update({"_id": {field: '$' + field for field in fields} or {field: (value if not isinstance(value, str) or value.startswith('$') else '$' + value) for field, value in kfields.items()}})
    def sort_documents(self, *fields, descending=False):
        self.append(self._sort.descending_order(*fields) if descending else self._sort.ascending_order(*fields))

class WebFilterBlockedPipeline(BasePipeline):
    def aggregate_root(self, *args):
        self.append(self._reroot.reset_root(*args))
    def aggregate_data(self, **kfields):
        self.group.sum_metrics(**kfields)
        self.aggregate_root(*kfields)
    @classmethod
    def create(cls, data, devices):
        pipeline = cls(data, devices)
        pipeline.match.equals(action="blocked")
        pipeline.set_composite_group("url", "source_ip", "profile")
        pipeline.aggregate_data(count=1)
        pipeline.sort_documents("count", descending=True)
        pipeline.set_simple_group(field="url")
        pipeline.group.sum_metrics(count="count")
        pipeline.group.nest_field("source_ip", "profile", "count", key="users")
        pipeline.sort_documents("count", descending=True)
        return pipeline

Example output of WebFilterBlockedPipeline.create() in JSON-format:

[{
    "$match": {
        "$and": [{
            "time": {
                "$gte": "2020-07-15T16:04:19",
                "$lt": "2020-07-16T16:04:19"
            }
        },
        {
            "device": {
                "$in": ["FG100ETK18035573"]
            }
        },
        {
            "action": {
                "$eq": "blocked"
            }
        }]
    }
},
{
    "$group": {
        "_id": {
            "url": "$url",
            "source_ip": "$source_ip",
            "profile": "$profile"
        },
        "count": {
            "$sum": 1
        }
    }
},
{
    "$replaceRoot": {
        "newRoot": {
            "$mergeObjects": [{
                "count": "$count"
            },
            "$_id"]
        }
    }
},
{
    "$sort": {
        "count": -1
    }
},
{
    "$group": {
        "_id": "$url",
        "count": {
            "$sum": {
                "$toLong": "$count"
            }
        },
        "users": {
            "$push": {
                "source_ip": "$source_ip",
                "profile": "$profile",
                "count": "$count"
            }
        }
    }
},
{
    "$addFields": {
        "id": "$_id"
    }
},
{
    "$project": {
        "_id": 0
    }
},
{
    "$sort": {
        "count": -1
    }
}]

I’d like to ask if this is an acceptable implementation and, if not, if I should just scratch it completely, follow another design pattern or any constructive criticism.