Flask + PrintedSimply

# app.py
import os
import requests
from dotenv import load_dotenv
from flask import Flask, request, jsonify

from webhooks import webhooks

load_dotenv()
app = Flask(__name__)
app.register_blueprint(webhooks)

# No SDK required — call the REST API with the requests library.
BASE_URL = "https://portal.printedsimply.com/api/v1"
HEADERS = {
    "x-api-key": os.environ["PRINTEDSIMPLY_API_KEY"],
    "Content-Type": "application/json",
}


@app.route("/products")
def list_products():
    resp = requests.get(f"{BASE_URL}/products", headers=HEADERS)
    # Responses are wrapped as {"status", "data", "request_id"}.
    return jsonify(resp.json()["data"]["products"])


@app.route("/estimate", methods=["POST"])
def estimate():
    data = request.json
    resp = requests.post(
        f"{BASE_URL}/pricing/estimate",
        headers=HEADERS,
        json={
            "items": data["items"],
            "shipping_method": data.get("shipping_method", "standard"),
        },
    )
    return jsonify(resp.json()["data"])


@app.route("/orders", methods=["POST"])
def create_order():
    data = request.json
    # reference is your own order id. It is stamped on the order, and it makes
    # the Idempotency-Key: a retry of the same order returns the first result
    # instead of placing a second one.
    reference = data.get("reference")
    if not reference:
        return jsonify({"error": "reference (your order id) is required"}), 400
    resp = requests.post(
        f"{BASE_URL}/order",
        headers={**HEADERS, "Idempotency-Key": f"order-{reference}"},
        json={
            "reference": reference,
            "items": data["items"],
            "address_to": data["address_to"],
            "shipping": {"method": data.get("shipping_method", "standard")},
        },
    )
    if resp.status_code >= 400:
        return jsonify(resp.json().get("error", {"message": "Order failed"})), resp.status_code
    return jsonify(resp.json()["data"])


@app.route("/orders/<order_id>")
def get_order(order_id):
    resp = requests.get(f"{BASE_URL}/order/{order_id}", headers=HEADERS)
    return jsonify(resp.json()["data"])

# webhooks.py
import json
import os

from flask import Blueprint, request

import hmac
import hashlib
import time

TOLERANCE_SECONDS = 300

# header is the raw X-PS-Signature-V2 value, e.g. "t=1754222400,v1=abc123..."
# During a secret rotation the header carries one v1 per live secret
# ("t=...,v1=<new>,v1=<old>"), so collect them all and accept ANY match.
# A missing header (None) is refused like a malformed one.
def verify_webhook(raw_body: bytes, header: str, secret: str) -> bool:
    t, v1s = "", []
    for part in (header or "").split(","):
        if "=" not in part:
            continue
        key, value = part.strip().split("=", 1)
        if key == "t":
            t = value
        elif key == "v1":
            v1s.append(value)
    # A Unix timestamp is ten digits. Refusing a longer one also keeps int()
    # from raising on a header padded past Python's digit limit.
    if not (t.isascii() and t.isdigit()) or len(t) > 12 or not v1s:
        return False

    # Reject a captured request replayed outside the window. This check is the
    # reason to prefer V2 — a body-only signature never expires.
    if abs(int(time.time()) - int(t)) > TOLERANCE_SECONDS:
        return False

    # The timestamp is signed along with the body, so it cannot be swapped out.
    expected = hmac.new(
        secret.encode(),
        t.encode() + b"." + raw_body,
        hashlib.sha256,
    ).hexdigest()
    # Compared as bytes: compare_digest raises on a str with non-ASCII characters.
    return any(hmac.compare_digest(v1.encode(), expected.encode()) for v1 in v1s)


webhooks = Blueprint("webhooks", __name__)

# Delivery is at-least-once: a retry or a redelivery sends the same event
# again, with the same X-PS-Event-Id. Skip an id you have already handled.
# In production, keep handled ids in your database under a unique index;
# this set stands in for it.
handled = set()


@webhooks.route("/webhooks/printedsimply", methods=["POST"])
def printedsimply_webhook():
    # The raw bytes, exactly as signed — not request.json.
    raw_body = request.get_data()
    signature = request.headers.get("X-PS-Signature-V2")
    if not verify_webhook(raw_body, signature, os.environ["PRINTEDSIMPLY_WEBHOOK_SECRET"]):
        return "Invalid signature", 401

    event_id = request.headers.get("X-PS-Event-Id", "")
    if event_id in handled:
        return "OK", 200

    # Every delivery is {"id", "event", "timestamp", "data"}; the order is in data.
    event = json.loads(raw_body)
    order = event.get("data", {})
    if event.get("event") == "order.shipped":
        tracking_number = order.get("tracking_number")
        print(f"Shipped! Order {order.get('order_id')}, tracking {tracking_number}")
        # Update your database
    elif event.get("event") == "order.delivered":
        print(f"Delivered! Order {order.get('order_id')}")
    # Answer 200 to events you do not handle too: anything else is a failed delivery.

    handled.add(event_id)
    return "OK", 200