import base64 from marshmallow import ( Schema, fields, validate, validates, ValidationError, validates_schema, INCLUDE, ) from core_notifications import models from core_notifications.api.fields import EmailOrEmailsListField from core_notifications.consts import AttachmentTypeFile class Attachment(Schema): filename = fields.Str(required=True) filetype = fields.Str( validate=validate.OneOf([item.value for item in AttachmentTypeFile]), required=True, ) base64_data = fields.Str(required=True) @validates("base64_data") def validate_base64_data(self, value): try: base64.b64decode(value) except Exception: raise ValidationError("Error decoding base64 data") class Template(Schema): id = fields.Integer(required=True) context = fields.Dict(required=False) @validates("id") def validate_id(self, value): template = models.Template.query.get(value) if template is None: raise ValidationError("The template does not exist") if not template.is_active: raise ValidationError("The template is not active") class EmailSendRequestBodyItemSchema(Schema): to = EmailOrEmailsListField(fields.Email(), required=True) subject = fields.Str(required=False) template = fields.Nested(Template(), required=False) body = fields.Str(required=False) cc = EmailOrEmailsListField( fields.Email(), required=False, allow_none=True ) attachments = fields.List(fields.Nested(Attachment(), required=False)) campaign_id = fields.Str(required=False) track_clicks = fields.Bool(required=False) @validates_schema def validate_template_or_body(self, data, **kwargs): if not data.get("body") and not data.get("template"): raise ValidationError("Body or template is required") @validates_schema def validate_template_or_subject(self, data, **kwargs): if not data.get("subject") and not data.get("template"): raise ValidationError("Subject or template is required") class EmailSendResponse(Schema): class Meta: unknown = INCLUDE class StatusResponse(Schema): status = fields.String()