Kod örnekleri: Python

Gereksinimler: Python 3.9 ya da sonrası; yalnızca standart kitaplık.

"""UPPU API v1: a minimal client (Python 3.9+, standard library only).

Configuration comes from the environment, never from the source code:
  UPPU_API_BASE  https://messaging-staging.uppu.net/v1 (production: https://messaging.uppu.net/v1)
  UPPU_API_KEY   uppu_test_... on staging, uppu_live_... in production. Never log it.
"""

import json
import os
import time
import urllib.error
import urllib.request

BASE = os.environ.get("UPPU_API_BASE", "https://messaging-staging.uppu.net/v1").rstrip("/")


def uppu_request(method, path, body=None, idempotency_key=None, attempts=3):
    """Call the API; return (status, decoded body or None, headers).

    Network errors, 5xx answers and errors marked retryable are retried with
    the same Idempotency-Key, so a retry never sends a message twice.
    """
    headers = {"Authorization": "Bearer " + os.environ["UPPU_API_KEY"], "Accept": "application/json"}
    data = None

    if body is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")

    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    attempt = 0

    while True:
        attempt += 1
        request = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)

        try:
            with urllib.request.urlopen(request, timeout=20) as response:
                status, raw, response_headers = response.status, response.read(), response.headers
        except urllib.error.HTTPError as error:
            status, raw, response_headers = error.code, error.read(), error.headers
        except OSError:
            # Every network error: URLError, a timeout while waiting for the
            # answer (socket.timeout on Python 3.9), a reset connection
            if attempt >= attempts:
                raise
            time.sleep(attempt)
            continue

        decoded = json.loads(raw) if raw else None
        retryable = status >= 500 or bool(((decoded or {}).get("error") or {}).get("retryable"))

        if not retryable or attempt >= attempts:
            return status, decoded, response_headers

        # Retry-After when UPPU sends it, otherwise 1, 2, ... seconds; at most 10
        retry_after = response_headers.get("Retry-After") or ""
        time.sleep(min(int(retry_after) if retry_after.isdigit() else attempt, 10))


def uppu_error_code(status, body):
    """The error code of an answer, for logs and decisions (never the message)."""
    return ((body or {}).get("error") or {}).get("code") or "http_%d" % status

"""Send a text message: POST /v1/messages, scope messages:send."""

from uppu_client import uppu_error_code, uppu_request

channel_id = "01j9z0ch5k2m4n6p8q0r2s4t6v"  # from GET /v1/channels
contact_id = "01j8x7ct5k2m4n6p8q0r2s4t6v"  # from message.received: data.contact.id

status, message, _ = uppu_request(
    "POST",
    "/messages",
    {
        "channel_id": channel_id,
        "to": {"contact_id": contact_id},
        "content": {"type": "text", "text": {"body": "Your order has shipped."}},
        "client_reference": "order-1001",
    },
    idempotency_key="order-1001-shipped",  # the same key for every retry of this send
)

if status in (200, 202):
    # 202: queued; 200: a replay of the same send (Idempotent-Replayed: true)
    print("Message", message["id"], "is", message["status"])
else:
    # For example window_closed: the 24-hour window is closed, send a template instead
    print("Not sent:", uppu_error_code(status, message))

"""Send an approved WhatsApp template, also outside the 24-hour window.

POST /v1/messages, scope messages:send. GET /v1/templates shows each
template's parameters and a ready send_example.
"""

from uppu_client import uppu_error_code, uppu_request

status, message, _ = uppu_request(
    "POST",
    "/messages",
    {
        "channel_id": "01j9z0ch5k2m4n6p8q0r2s4t6v",
        "to": {"contact_id": "01j8x7ct5k2m4n6p8q0r2s4t6v"},
        "content": {
            "type": "template",
            "template": {
                "name": "uppu_sim_order_update",  # a template of the staging simulator
                "language": "tr",
                "body": {"parameters": ["Deniz", "A-1001"]},  # {{1}} and {{2}}
            },
        },
        "client_reference": "order-A-1001",
    },
    idempotency_key="order-A-1001-shipped",
)

if status in (200, 202):
    print("Template message", message["id"], "is", message["status"])
else:
    # For example consent_required: a MARKETING template needs a recorded consent
    print("Not sent:", uppu_error_code(status, message))

"""List messages page by page: GET /v1/messages, scope messages:read.

Use it to reconcile after a webhook outage.
"""

import urllib.parse

from uppu_client import uppu_error_code, uppu_request

query = {
    "channel_id": "01j9z0ch5k2m4n6p8q0r2s4t6v",
    "created_after": "2026-10-20T00:00:00Z",
    "order": "asc",
    "limit": 100,
}

while True:
    status, page, _ = uppu_request("GET", "/messages?" + urllib.parse.urlencode(query))

    if status != 200:
        raise RuntimeError("Listing failed: " + uppu_error_code(status, page))

    for message in page["data"]:
        print(message["id"], message["direction"], message["status"])

    # The cursor carries the order; repeat the filters on every page
    if page["next_cursor"] is None:
        break

    query["cursor"] = page["next_cursor"]

"""Mark a message of the person as read (WhatsApp only).

POST /v1/messages/{id}/read, scope messages:send. No Idempotency-Key is
needed: a repeat answers 200.
"""

import urllib.parse

from uppu_client import uppu_error_code, uppu_request

message_id = "01jab3t6ms2m4n6p8q0r2s4t6v"  # from message.received: data.message.id

status, receipt, _ = uppu_request("POST", "/messages/" + urllib.parse.quote(message_id) + "/read", {})

if status in (200, 202):
    print("Read receipt", receipt["state"])
elif uppu_error_code(status, receipt) == "not_inbound":
    # Nothing to mark (your own message, a reaction, a notice): final, do not retry
    print("Nothing to mark:", receipt["error"]["details"].get("reason", "unknown"))
else:
    print("Not marked:", uppu_error_code(status, receipt))

"""Receive UPPU webhooks (Python 3.9+, standard library only).

Check the signature over the raw body, drop repeats by webhook-id, answer
2xx at once and process the event later. The signing secret comes from the
environment: UPPU_WEBHOOK_SECRET, and during a secret rotation the previous
one in UPPU_WEBHOOK_SECRET_PREVIOUS. Run the receiver under any WSGI server,
for example: gunicorn verify_webhook:application
"""

import base64
import binascii
import hashlib
import hmac
import json
import os
import time


def verify_uppu_signature(secrets, headers, raw_body, now=None, tolerance=300):
    """Standard Webhooks v1 check: base64(HMAC-SHA256(key, id.timestamp.body)).

    secrets: the current secret and, during a rotation, the previous one ("whsec_...").
    headers: lower-case header names; raw_body: the body exactly as received (bytes).
    """
    webhook_id = headers.get("webhook-id") or ""
    timestamp = headers.get("webhook-timestamp") or ""
    header = headers.get("webhook-signature") or ""
    current = time.time() if now is None else now

    if not webhook_id or not (timestamp.isascii() and timestamp.isdigit()) or abs(current - int(timestamp)) > tolerance:
        return False

    signed = webhook_id.encode() + b"." + timestamp.encode() + b"." + raw_body

    for secret in secrets:
        if not secret or not secret.startswith("whsec_"):
            continue

        try:
            key = base64.b64decode(secret[6:], validate=True)
        except (binascii.Error, ValueError):
            continue

        if not key:
            continue

        expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest())

        for part in header.split(" "):
            version, _, signature = part.partition(",")

            if version == "v1" and hmac.compare_digest(expected, signature.encode()):
                return True

    return False


def application(environ, start_response):
    """A minimal WSGI receiver."""
    length = int(environ.get("CONTENT_LENGTH") or 0)
    body = environ["wsgi.input"].read(length)
    headers = {
        "webhook-id": environ.get("HTTP_WEBHOOK_ID", ""),
        "webhook-timestamp": environ.get("HTTP_WEBHOOK_TIMESTAMP", ""),
        "webhook-signature": environ.get("HTTP_WEBHOOK_SIGNATURE", ""),
    }
    secrets = [os.environ.get("UPPU_WEBHOOK_SECRET", ""), os.environ.get("UPPU_WEBHOOK_SECRET_PREVIOUS", "")]

    if environ.get("REQUEST_METHOD") != "POST" or not verify_uppu_signature(secrets, headers, body):
        # Anything but 2xx: the wrong-signature probe expects it
        start_response("401 Unauthorized", [("Content-Type", "text/plain")])
        return [b"invalid signature"]

    event = json.loads(body)

    # Test events of the panel and repeats: answer 2xx and do nothing else
    if not event.get("test") and store_event_once(event["id"], event["type"], body):
        enqueue(event)  # process outside the request: answer within 10 seconds

    start_response("204 No Content", [])
    return []


def store_event_once(event_id, event_type, body):
    """Your storage: True when the id is new. Keep ids for at least 35 days."""
    return True


def enqueue(event):
    """Hand the event to your queue or background worker."""
    if event["type"] == "message.received":
        pass  # data.message, data.contact and data.service_window
    elif event["type"] == "message.status":
        pass  # data.message.id or data.message.client_reference, data.status
    elif event["type"] == "contact.erased":
        pass  # delete your copy of data.contact_id
    # ignore event types you do not know