""" Proveedor SendGrid. Traduce un `EmailPayload` al formato de la API v3 de SendGrid y envia el mensaje. Cubre lo que necesita un servicio transaccional: multiples destinatarios, copia y copia oculta, adjuntos, imagenes incrustadas, plantillas dinamicas, etiquetas y modo de pruebas. Un detalle importante para la trazabilidad: se envia el identificador interno del envio en `custom_args`. SendGrid lo devuelve tal cual en cada evento del webhook, lo que permite enlazar "este correo reboto" con la fila exacta de la auditoria sin depender de heuristicas sobre el message-id. """ from __future__ import annotations import base64 import logging from django.conf import settings from python_http_client.exceptions import HTTPError from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import ( Attachment as SGAttachment, ) from sendgrid.helpers.mail import ( Bcc, Category, Cc, ContentId, Disposition, Email, FileContent, FileName, FileType, Header, Mail, MailSettings, Personalization, ReplyTo, SandBoxMode, To, ) from notifications.channels.email.base import ( EmailPayload, EmailProvider, ProviderConfigurationError, SendResult, is_retryable_status, ) logger = logging.getLogger("notifications.providers.sendgrid") #: Clave con la que viaja el identificador interno del envio. CUSTOM_ARG_MESSAGE_ID = "api_message_id" class SendGridProvider(EmailProvider): name = "sendgrid" supports_templates = True def __init__(self, api_key: str | None = None, *, sandbox: bool | None = None): self.api_key = api_key if api_key is not None else settings.SENDGRID_API_KEY self.sandbox = ( sandbox if sandbox is not None else settings.SENDGRID_SANDBOX_MODE ) self.timeout = settings.SENDGRID_TIMEOUT_SECONDS def check_configuration(self) -> None: if not self.api_key: raise ProviderConfigurationError( "SENDGRID_API_KEY no esta configurada." ) # -- Construccion del mensaje ------------------------------------------ def _build_mail(self, payload: EmailPayload) -> Mail: mail = Mail() mail.from_email = Email(payload.from_email, payload.from_name or None) # Una unica personalizacion con todos los destinatarios: SendGrid # entrega un correo por cada direccion de `to` pero comparte el # cuerpo, que es el comportamiento esperado de un envio normal. personalization = Personalization() for address in payload.to: personalization.add_to(To(address)) for address in payload.cc: personalization.add_cc(Cc(address)) for address in payload.bcc: personalization.add_bcc(Bcc(address)) if payload.template_id and payload.template_data: personalization.dynamic_template_data = dict(payload.template_data) mail.add_personalization(personalization) if payload.reply_to: mail.reply_to = ReplyTo(payload.reply_to) if payload.template_id: # Con plantilla, el asunto y el cuerpo los define la plantilla. mail.template_id = payload.template_id else: mail.subject = payload.subject # El orden importa: en un mensaje multipart/alternative la ultima # parte es la que el cliente de correo prefiere mostrar, asi que # el texto plano va primero y el HTML despues. if payload.text: mail.add_content(_content("text/plain", payload.text)) if payload.html: mail.add_content(_content("text/html", payload.html)) for attachment in payload.attachments: mail.add_attachment(_to_sendgrid_attachment(attachment)) for category in payload.categories[:10]: # SendGrid admite 10 como maximo mail.add_category(Category(category)) for key, value in payload.headers.items(): mail.add_header(Header(key, value)) for key, value in payload.custom_args.items(): mail.add_custom_arg(_custom_arg(key, value)) if self.sandbox: # En modo de pruebas SendGrid valida la peticion completa pero no # entrega nada. Util para entornos de QA. mail_settings = MailSettings() mail_settings.sandbox_mode = SandBoxMode(True) mail.mail_settings = mail_settings return mail # -- Envio -------------------------------------------------------------- def send(self, payload: EmailPayload) -> SendResult: self.check_configuration() try: mail = self._build_mail(payload) except Exception as exc: # noqa: BLE001 - error al armar el mensaje logger.exception("No se pudo construir el mensaje para SendGrid") return SendResult.failure( self.name, error_code="build_error", error_message=str(exc), retryable=False, ) client = SendGridAPIClient(self.api_key) try: # Se usa el cliente HTTP interno en lugar de `client.send()` # porque este ultimo no permite fijar un tiempo de espera, y una # peticion sin timeout puede dejar bloqueado al worker. response = client.client.mail.send.post( request_body=mail.get(), timeout=self.timeout ) except HTTPError as exc: status_code = getattr(exc, "status_code", None) body = _decode(getattr(exc, "body", b"")) logger.warning( "SendGrid rechazo el envio (HTTP %s)", status_code, extra={"status_code": status_code, "provider": self.name}, ) return SendResult.failure( self.name, error_code=f"http_{status_code}", error_message=body or str(exc), status_code=status_code, retryable=is_retryable_status(status_code), ) except Exception as exc: # noqa: BLE001 - red, DNS, TLS, timeout logger.warning( "Fallo de red al contactar con SendGrid: %s", exc.__class__.__name__ ) return SendResult.failure( self.name, error_code="network_error", error_message=f"{exc.__class__.__name__}: {exc}", retryable=True, ) status_code = getattr(response, "status_code", None) message_id = _header(response, "X-Message-Id") if status_code is not None and 200 <= status_code < 300: logger.info( "Correo aceptado por SendGrid", extra={ "provider": self.name, "status_code": status_code, "provider_message_id": message_id, "recipients": len(payload.all_recipients), }, ) return SendResult.ok(self.name, message_id, status_code) return SendResult.failure( self.name, error_code=f"http_{status_code}", error_message=_decode(getattr(response, "body", b"")), status_code=status_code, retryable=is_retryable_status(status_code), ) # --------------------------------------------------------------------------- # Ayudas # --------------------------------------------------------------------------- def _content(mime_type: str, value: str): from sendgrid.helpers.mail import Content return Content(mime_type, value) def _custom_arg(key: str, value: str): from sendgrid.helpers.mail import CustomArg return CustomArg(key, value) def _to_sendgrid_attachment(attachment) -> SGAttachment: """Convierte un `Attachment` propio al formato de SendGrid.""" sg_attachment = SGAttachment() sg_attachment.file_content = FileContent( base64.b64encode(attachment.content).decode("ascii") ) sg_attachment.file_name = FileName(attachment.filename) sg_attachment.file_type = FileType(attachment.content_type) if attachment.inline and attachment.content_id: sg_attachment.disposition = Disposition("inline") sg_attachment.content_id = ContentId(attachment.content_id) else: sg_attachment.disposition = Disposition("attachment") return sg_attachment def _header(response, name: str) -> str: """Lee una cabecera de la respuesta sin importar su representacion.""" headers = getattr(response, "headers", None) if headers is None: return "" try: value = headers.get(name) except AttributeError: return "" return value or "" def _decode(body) -> str: if isinstance(body, bytes): return body.decode("utf-8", errors="replace") return str(body or "")