from apollo_utils.service.schemas.fields.enum import EnumField, EnumListField from marshmallow import ValidationError, fields, pre_dump, validate, validates_schema from server.constants.core import CheckUniqueMode from server.constants.messages import MESSAGE_DATA_MAX_SIZE, MESSAGE_DEFAULT_TTL, MESSAGE_MAX_TTL, \ MESSAGE_META_MAX_SIZE, MessageInclude from server.schemas.messages.fields import JsonField class TTLMixin: ttl = fields.Int(missing=MESSAGE_DEFAULT_TTL, validate=validate.Range(min=0, max=MESSAGE_MAX_TTL)) class MessageIncludeMixin: include = EnumListField(MessageInclude, missing=[], allow_all=True) class MetaDataNoneMixin: meta = JsonField(missing=None, default=None, max_size=MESSAGE_META_MAX_SIZE, return_orig_data=True) data = JsonField(missing=None, default=None, max_size=MESSAGE_DATA_MAX_SIZE, return_orig_data=True) class ContextNoneMixin: context = fields.List(fields.Dict(), default=None) class PublicMixin: public = fields.Bool(missing=True) class CheckUniqueMixin: unique_mode = EnumField(CheckUniqueMode, missing=CheckUniqueMode.INACTIVE) unique_key = fields.Str(missing=None) @validates_schema def validate_check_unique_parameters(self, data, **kwargs): unique_mode, unique_key = data.get("unique_mode"), data.get("unique_key") if bool(unique_mode) ^ bool(unique_key): raise ValidationError( f"'unique_mode' and 'unique_key' should be both set or reset.\n" f"Got unique_mode={unique_mode}, unique_key={unique_key}" ) class CodeMixin: code = fields.String(required=True, validate=validate.Length(max=50)) class EventIdNoneMixin: event_id = fields.Int(missing=None, default=None, required=False, validate=validate.Range(min=0)) class MessageIdNoneMixin: message_id = fields.Int(missing=None, default=None, required=False, validate=validate.Range(min=0)) class ExternalIdNoneMixin: external_id = fields.Str(missing=None, default=None, required=False, validate=validate.Length(min=0, max=80)) class AccountIdNoneMixin: account_id = fields.Int(missing=None, default=None, required=False, validate=validate.Range(min=0)) class DeviceIdNoneMixin: device_id = fields.Int(missing=None, default=None, required=False, validate=validate.Range(min=0)) class MessageRelationMixin(EventIdNoneMixin, MessageIdNoneMixin, AccountIdNoneMixin, ExternalIdNoneMixin): pass class ListToDataDumpMixin: @pre_dump def prepare_list(self, data, **kwargs): if data is not None and isinstance(data, list): return {"data": data} return data