clearml-server/apiserver/apimodels/base.py

from jsonmodels import models, fields
from jsonmodels.validators import Length
from mongoengine.base import BaseDocument

from apiserver.apimodels import DictField, ListField


class MongoengineFieldsDict(DictField):
    """
    DictField representing mongoengine field names/value mapping.
    Used to convert mongoengine-style field/subfield notation to user-presentable syntax, including handling update
        operators.
    """

    mongoengine_update_operators = (
        "inc",
        "dec",
        "push",
        "push_all",
        "pop",
        "pull",
        "pull_all",
        "add_to_set",
    )

    @staticmethod
    def _normalize_mongo_value(value):
        if isinstance(value, BaseDocument):
            return value.to_mongo()
        return value

    @classmethod
    def _normalize_mongo_field_path(cls, path, value):
        parts = path.split("__")
        if len(parts) > 1:
            if parts[0] == "set":
                parts = parts[1:]
            elif parts[0] == "unset":
                parts = parts[1:]
                value = None
            elif parts[0] in cls.mongoengine_update_operators:
                return None, None
        return ".".join(parts), cls._normalize_mongo_value(value)

    def parse_value(self, value):
        value = super(MongoengineFieldsDict, self).parse_value(value)
        return {
            k: v
            for k, v in (self._normalize_mongo_field_path(*p) for p in value.items())
            if k is not None
        }


class UpdateResponse(models.Base):
    updated = fields.IntField(required=True)
    fields = MongoengineFieldsDict()


class PagedRequest(models.Base):
    page = fields.IntField()
    page_size = fields.IntField()


class IdResponse(models.Base):
    id = fields.StringField(required=True)


class MakePublicRequest(models.Base):
    ids = ListField(items_types=str, validators=[Length(minimum_value=1)])