infrasynth-backend-kit/infrasynth/billing/serializers.py
jcv-dev 551b42eab5 feat: production-hardening pass across the kit
Close the gaps between the documented contract (API-STANDARD, TENANCY,
ENTITLEMENTS) and the implementation, and remove committed build artifacts.

Security:
- verify + process inbound webhooks (HMAC/handler verify, size limit,
  timestamp tolerance, idempotency via InboundEvent.external_id)
- real 2FA login flow (pre-auth challenge; tokens only after verify/recovery)
- wire HybridPermission into security/audit views; add API-key rotate and
  users/<id>/permissions|roles endpoints
- tenant-scoped throttling on by default; webhook replay protection
- verify MercadoPago webhook signatures
- login brute-force guard, configurable password policy, real ALTCHA PoW

Correctness:
- apply verified billing webhooks idempotently (subscription/entitlement/
  invoice/PaymentTransaction); scheduled payment lifecycle jobs
- capture audit update diffs automatically; add audit retention purge
- working notification retries, per-channel rate limits, log retention
- pluggable virus scanner, upload-size limit, pipeline toggle
- feature rollout %/environment targeting; settings-driven registrations
- workflow guards (instance cap, route depth, self-assignment, clone on re-entry)
- wire every previously-dead INFRASYNTH_* setting; drop truly dead ones

Delivery:
- README + CHANGELOG; CI format check + coverage gate
- keep test media out of the tree; untrack .coverage, __pycache__,
  egg-info, docs/ and invoice artifacts
2026-09-24 10:41:21 -05:00

236 lines
6.4 KiB
Python

from rest_framework import serializers
from .models import (
App,
Entitlement,
Invoice,
PaymentGateway,
PaymentTransaction,
Plan,
Subscription,
)
class AppSerializer(serializers.ModelSerializer):
class Meta:
model = App
fields = ["id", "slug", "name", "monetization", "is_active", "metadata"]
read_only_fields = ["id"]
class PlanSerializer(serializers.ModelSerializer):
app_info = serializers.SerializerMethodField()
gateway_info = serializers.SerializerMethodField()
class Meta:
model = Plan
fields = [
"id",
"app",
"app_info",
"slug",
"name",
"price_amount",
"price_currency",
"interval",
"trial_days",
"features",
"limits",
"is_active",
"gateway",
"gateway_info",
"external_id",
]
read_only_fields = ["id"]
def get_app_info(self, obj):
return {"id": obj.app_id, "slug": obj.app.slug, "name": obj.app.name}
def get_gateway_info(self, obj):
if obj.gateway_id:
return {"slug": obj.gateway_id, "display_name": obj.gateway.display_name}
return None
class EntitlementSerializer(serializers.ModelSerializer):
app_info = serializers.SerializerMethodField()
plan_info = serializers.SerializerMethodField()
class Meta:
model = Entitlement
fields = [
"id",
"app",
"app_info",
"plan",
"plan_info",
"status",
"started_at",
"current_period_end",
"expires_at",
"cancel_at_period_end",
"source",
"metadata",
]
read_only_fields = ["id", "started_at"]
def get_app_info(self, obj):
return {"slug": obj.app.slug, "name": obj.app.name}
def get_plan_info(self, obj):
if obj.plan_id:
return {"id": obj.plan_id, "slug": obj.plan.slug, "name": obj.plan.name}
return None
class PaymentGatewaySerializer(serializers.ModelSerializer):
class Meta:
model = PaymentGateway
fields = [
"slug",
"display_name",
"gateway_class",
"config",
"is_active",
"supported_currencies",
"webhook_secret",
]
read_only_fields = ["slug"]
class SubscriptionSerializer(serializers.ModelSerializer):
plan_info = serializers.SerializerMethodField()
gateway_info = serializers.SerializerMethodField()
entitlement_info = serializers.SerializerMethodField()
class Meta:
model = Subscription
fields = [
"id",
"entitlement",
"entitlement_info",
"plan",
"plan_info",
"gateway",
"gateway_info",
"external_id",
"status",
"current_period_start",
"current_period_end",
"cancel_at_period_end",
"cancelled_at",
"trial_end",
"metadata",
]
read_only_fields = ["id"]
def get_plan_info(self, obj):
if obj.plan_id:
return {"id": obj.plan_id, "name": obj.plan.name, "slug": obj.plan.slug}
return None
def get_gateway_info(self, obj):
if obj.gateway_id:
return {"slug": obj.gateway_id, "display_name": obj.gateway.display_name}
return None
def get_entitlement_info(self, obj):
if obj.entitlement_id:
return {"id": obj.entitlement_id, "status": obj.entitlement.status}
return None
class InvoiceSerializer(serializers.ModelSerializer):
subscription_info = serializers.SerializerMethodField()
gateway_info = serializers.SerializerMethodField()
pdf_file_info = serializers.SerializerMethodField()
class Meta:
model = Invoice
fields = [
"id",
"subscription",
"subscription_info",
"gateway",
"gateway_info",
"external_id",
"invoice_number",
"amount",
"currency",
"tax_amount",
"tax_name",
"status",
"due_date",
"paid_at",
"line_items",
"pdf_file",
"pdf_file_info",
"metadata",
]
read_only_fields = ["id"]
def get_subscription_info(self, obj):
if obj.subscription_id:
return {"id": obj.subscription_id}
return None
def get_gateway_info(self, obj):
if obj.gateway_id:
return {"slug": obj.gateway_id, "display_name": obj.gateway.display_name}
return None
def get_pdf_file_info(self, obj):
if obj.pdf_file_id:
return {"id": obj.pdf_file_id, "filename": obj.pdf_file.original_filename}
return None
class PaymentTransactionSerializer(serializers.ModelSerializer):
invoice_info = serializers.SerializerMethodField()
gateway_info = serializers.SerializerMethodField()
class Meta:
model = PaymentTransaction
fields = [
"id",
"invoice",
"invoice_info",
"gateway",
"gateway_info",
"external_id",
"amount",
"currency",
"status",
"payment_method",
"metadata",
"created_at",
]
read_only_fields = ["id", "created_at"]
def get_invoice_info(self, obj):
if obj.invoice_id:
return {
"id": obj.invoice_id,
"invoice_number": obj.invoice.invoice_number,
}
return None
def get_gateway_info(self, obj):
if obj.gateway_id:
return {"slug": obj.gateway_id, "display_name": obj.gateway.display_name}
return None
class CheckoutSerializer(serializers.Serializer):
"""``POST /billing/checkout/`` — tenant comes from the token, never the body."""
app = serializers.SlugField()
plan = serializers.SlugField()
success_url = serializers.URLField(required=False)
cancel_url = serializers.URLField(required=False)
class SubscribeSerializer(serializers.Serializer):
plan_slug = serializers.SlugField()
success_url = serializers.URLField(required=False)
cancel_url = serializers.URLField(required=False)