7af9b2c82c
- New billing/paddle.py: Paddle-specific functions (build_checkout_payload, cancel_subscription, get_management_url, verify_webhook, parse_webhook) - routes.py: _provider() dispatch function selects paddle or stripe module - Checkout/manage/cancel routes now delegate to _provider() - /webhook/paddle always active (existing subscribers) - /webhook/stripe endpoint added (returns 404 until Stripe configured) - Shared _handle_webhook_event() processes normalized events from any provider - _price_id_to_key() queries payment_products with paddle_products fallback Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""
|
|
Paddle payment provider — checkout, webhook verification, subscription management.
|
|
|
|
Exports the 5 functions that billing/routes.py dispatches to:
|
|
- build_checkout_payload()
|
|
- build_multi_item_checkout_payload()
|
|
- cancel_subscription()
|
|
- get_management_url()
|
|
- handle_webhook()
|
|
"""
|
|
|
|
import json
|
|
|
|
from paddle_billing import Client as PaddleClient
|
|
from paddle_billing import Environment, Options
|
|
from paddle_billing.Notifications import Secret, Verifier
|
|
|
|
from ..core import config
|
|
|
|
|
|
def _paddle_client() -> PaddleClient:
|
|
"""Create a Paddle SDK client."""
|
|
env = Environment.SANDBOX if config.PADDLE_ENVIRONMENT == "sandbox" else Environment.PRODUCTION
|
|
return PaddleClient(config.PADDLE_API_KEY, options=Options(env))
|
|
|
|
|
|
class _WebhookRequest:
|
|
"""Minimal wrapper satisfying paddle_billing's Request Protocol."""
|
|
|
|
def __init__(self, body: bytes, headers):
|
|
self.body = body
|
|
self.headers = headers
|
|
|
|
|
|
_verifier = Verifier(maximum_variance=300)
|
|
|
|
|
|
def build_checkout_payload(
|
|
price_id: str, custom_data: dict, success_url: str,
|
|
) -> dict:
|
|
"""Build JSON payload for a single-item Paddle.js overlay checkout."""
|
|
return {
|
|
"items": [{"priceId": price_id, "quantity": 1}],
|
|
"customData": custom_data,
|
|
"settings": {"successUrl": success_url},
|
|
}
|
|
|
|
|
|
def build_multi_item_checkout_payload(
|
|
items: list[dict], custom_data: dict, success_url: str,
|
|
) -> dict:
|
|
"""Build JSON payload for a multi-item Paddle.js overlay checkout."""
|
|
return {
|
|
"items": items,
|
|
"customData": custom_data,
|
|
"settings": {"successUrl": success_url},
|
|
}
|
|
|
|
|
|
def cancel_subscription(provider_subscription_id: str) -> None:
|
|
"""Cancel a Paddle subscription at end of current billing period."""
|
|
from paddle_billing.Resources.Subscriptions.Operations import CancelSubscription
|
|
|
|
paddle = _paddle_client()
|
|
paddle.subscriptions.cancel(
|
|
provider_subscription_id,
|
|
CancelSubscription(effective_from="next_billing_period"),
|
|
)
|
|
|
|
|
|
def get_management_url(provider_subscription_id: str) -> str:
|
|
"""Get the Paddle customer portal URL for updating payment method."""
|
|
paddle = _paddle_client()
|
|
paddle_sub = paddle.subscriptions.get(provider_subscription_id)
|
|
return paddle_sub.management_urls.update_payment_method
|
|
|
|
|
|
def verify_webhook(payload: bytes, headers) -> bool:
|
|
"""Verify Paddle webhook signature. Returns True if valid or no secret configured."""
|
|
if not config.PADDLE_WEBHOOK_SECRET:
|
|
return True
|
|
try:
|
|
return _verifier.verify(
|
|
_WebhookRequest(payload, headers),
|
|
Secret(config.PADDLE_WEBHOOK_SECRET),
|
|
)
|
|
except (ConnectionRefusedError, ValueError):
|
|
return False
|
|
|
|
|
|
def parse_webhook(payload: bytes) -> dict:
|
|
"""Parse a Paddle webhook payload into a normalized event dict.
|
|
|
|
Returns dict with keys: event_type, subscription_id, customer_id,
|
|
user_id, supplier_id, plan, status, current_period_end, data, items.
|
|
"""
|
|
event = json.loads(payload)
|
|
event_type = event.get("event_type", "")
|
|
data = event.get("data") or {}
|
|
custom_data = data.get("custom_data") or {}
|
|
|
|
billing_period = data.get("current_billing_period") or {}
|
|
|
|
return {
|
|
"event_type": event_type,
|
|
"subscription_id": data.get("id", ""),
|
|
"customer_id": str(data.get("customer_id", "")),
|
|
"user_id": custom_data.get("user_id"),
|
|
"supplier_id": custom_data.get("supplier_id"),
|
|
"plan": custom_data.get("plan", ""),
|
|
"status": data.get("status", ""),
|
|
"current_period_end": billing_period.get("ends_at"),
|
|
"data": data,
|
|
"items": data.get("items", []),
|
|
"custom_data": custom_data,
|
|
}
|