Source code for sinch.domains.sms.sinch_events.v1.sms_sinch_event

import json
from typing import Any, Dict, Optional, Union

from pydantic import TypeAdapter

from sinch.domains.authentication.sinch_events.v1.authentication_validation import (
    validate_sinch_event_signature_with_nonce,
)
from sinch.domains.authentication.sinch_events.v1.sinch_event_utils import (
    decode_payload,
    normalize_iso_timestamp,
    parse_json,
)
from sinch.domains.sms.models.v1.response import (
    BatchDeliveryReport,
    RecipientDeliveryReport,
)
from sinch.domains.sms.sinch_events.v1.events.sms_sinch_event import (
    IncomingSMSSinchEvent,
    SmsSinchEventPayload,
)



[docs]
class SmsSinchEvent:
    def __init__(self, app_secret: Optional[str] = None):
        self.app_secret = app_secret


[docs]
    def validate_authentication_header(
        self,
        headers: Dict[str, str],
        json_payload: Union[str, bytes],
    ) -> bool:
        """
        Validate the authorization header for a callback request.

        :param headers: Incoming request's headers
        :type headers: Dict[str, str]
        :param json_payload: Incoming request's raw body (str or bytes)
        :type json_payload: Union[str, bytes]
        :returns: True if the X-Sinch-Webhook-Signature header is valid
        :rtype: bool
        """
        if not self.app_secret:
            return False
        payload_str = (
            decode_payload(json_payload, headers)
            if isinstance(json_payload, bytes)
            else json_payload
        )
        return validate_sinch_event_signature_with_nonce(
            self.app_secret, headers, payload_str
        )



[docs]
    def parse_event(
        self,
        event_body: Union[str, bytes, Dict[str, Any]],
        headers: Optional[Dict[str, str]] = None,
    ) -> SmsSinchEventPayload:
        """
        Parse the event payload into an SMS callback object.

        Handles datetime conversion for timestamp fields and routes to the
        appropriate event type based on the `type` field.

        :param event_body: The event payload (JSON string, raw bytes, or dict).
        :type event_body: Union[str, bytes, Dict[str, Any]]
        :param headers: Request headers (used to decode charset when event_body is bytes).
        :type headers: Optional[Dict[str, str]]
        :returns: A parsed SMS Sinch Event payload object.
        :rtype: SmsSinchEventPayload
        :raises ValueError: If the event type is unknown or parsing fails.
        """
        if isinstance(event_body, bytes):
            event_body = parse_json(decode_payload(event_body, headers))
        elif isinstance(event_body, str):
            event_body = parse_json(event_body)

        event_type = event_body.get("type")
        if not event_type:
            raise ValueError(f"Unknown SMS event: {json.dumps(event_body)}")

        # Handle delivery reports
        if event_type in ("delivery_report_sms", "delivery_report_mms"):
            return BatchDeliveryReport(**event_body)

        # Handle recipient delivery reports
        if event_type in (
            "recipient_delivery_report_sms",
            "recipient_delivery_report_mms",
        ):
            return RecipientDeliveryReport(**event_body)

        # Handle incoming SMS messages using discriminated union
        if event_type in ("mo_text", "mo_binary", "mo_media"):
            if "received_at" in event_body and isinstance(
                event_body["received_at"], str
            ):
                event_body["received_at"] = normalize_iso_timestamp(
                    event_body["received_at"]
                )
            if "sent_at" in event_body and isinstance(
                event_body["sent_at"], str
            ):
                event_body["sent_at"] = normalize_iso_timestamp(
                    event_body["sent_at"]
                )

            adapter = TypeAdapter(IncomingSMSSinchEvent)
            return adapter.validate_python(event_body)

        raise ValueError(f"Unknown SMS event type: {event_type}")