"""Durable provider webhook ingestion and subscription grant projection.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import hashlib from typing import Any, Mapping, Optional, Protocol from sqlalchemy import select from sqlalchemy.orm import Session from licensing_shared.catalog import LicensingCatalog, SkuKind from licensing_shared.constants import validate_identifier from .billing import ProviderBillingIncidentResolution from .errors import ServerErrorCode, ServerLicensingError from .models import ( AuditEvent, Grant, License, OutboxEvent, Subscription, SubscriptionItem, WebhookEvent, ) from .notifications import ( CustomerNotificationKind, billing_hold_notification_kind, enqueue_subscription_notification, subscription_transition_notification_kinds, ) from .security import new_identifier STRIPE_PROVIDER_ID = "stripe" STRIPE_SUBSCRIPTION_EVENT_TYPES = frozenset( ( "customer.subscription.created", "customer.subscription.updated", "customer.subscription.deleted", ) ) STRIPE_BILLING_INCIDENT_EVENT_TYPES = frozenset( ( "charge.dispute.closed", "charge.dispute.created", "charge.dispute.funds_reinstated", "charge.dispute.funds_withdrawn", "charge.dispute.updated", "charge.refunded", "refund.created", "refund.failed", "refund.updated", ) ) SUBSCRIPTION_STATUSES = frozenset( ( "incomplete", "incomplete_expired", "trialing", "active", "past_due", "paused", "canceled", "unpaid", ) ) ACTIVE_SUBSCRIPTION_STATUSES = frozenset(("trialing", "active", "past_due")) SUBSCRIPTION_OUTBOX_EVENT_TYPE = "stripe.subscription.project" BILLING_INCIDENT_OUTBOX_EVENT_TYPE = "stripe.billing_incident.project" MAXIMUM_OUTBOX_RETRY_ATTEMPTS = 100 class BillingIncidentProvider(Protocol): def resolve_billing_incident( self, incident: Mapping[str, object], ) -> ProviderBillingIncidentResolution: ... def _utc(value: datetime) -> datetime: if not isinstance(value, datetime): raise TypeError("value must be a datetime") if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _utc_from_unix(value: object, field_name: str) -> datetime: if isinstance(value, bool) or not isinstance(value, int): raise ValueError(f"{field_name} must be a Unix timestamp") try: return datetime.fromtimestamp(value, tz=timezone.utc) except (OSError, OverflowError, ValueError) as exc: raise ValueError(f"{field_name} is outside the supported timestamp range") from exc def normalize_stripe_subscription_event( event: Mapping[str, Any], price_sku_map: Mapping[str, str], ) -> dict[str, Any]: if not isinstance(event, Mapping): raise TypeError("event must be a mapping") if not isinstance(price_sku_map, Mapping): raise TypeError("price_sku_map must be a mapping") event_id = str(event.get("id") or "").strip() event_type = str(event.get("type") or "").strip() if not event_id or len(event_id) > 256: raise ValueError("Stripe event ID is empty or too long") if event_type not in STRIPE_SUBSCRIPTION_EVENT_TYPES: raise ValueError("Stripe event type is not supported") data = event.get("data") if not isinstance(data, Mapping) or not isinstance(data.get("object"), Mapping): raise ValueError("Stripe event data.object is missing") subscription = data["object"] subscription_id = str(subscription.get("id") or "").strip() customer_id = str(subscription.get("customer") or "").strip() status = str(subscription.get("status") or "").strip() if not subscription_id or len(subscription_id) > 256: raise ValueError("Stripe subscription ID is empty or too long") if not customer_id or len(customer_id) > 256: raise ValueError("Stripe customer ID is empty or too long") if status not in SUBSCRIPTION_STATUSES: raise ValueError("Stripe subscription status is unsupported") metadata = subscription.get("metadata") if not isinstance(metadata, Mapping): raise ValueError("Stripe subscription metadata is missing") license_id = validate_identifier(metadata.get("license_id"), "license_id") items_container = subscription.get("items") if not isinstance(items_container, Mapping): raise ValueError("Stripe subscription items are missing") item_values = items_container.get("data") if not isinstance(item_values, list) or not item_values or len(item_values) > 32: raise ValueError("Stripe subscription must contain between one and 32 items") normalized_items = [] starts = [] ends = [] seen_skus = set() for item in item_values: if not isinstance(item, Mapping): raise ValueError("Stripe subscription item must be an object") provider_item_id = str(item.get("id") or "").strip() price = item.get("price") if not isinstance(price, Mapping): raise ValueError("Stripe subscription item price is missing") price_id = str(price.get("id") or "").strip() sku_id = price_sku_map.get(price_id) if not isinstance(sku_id, str): raise ValueError(f"Stripe price is not mapped to a licensing SKU: {price_id}") normalized_sku = validate_identifier(sku_id, "sku_id") if normalized_sku in seen_skus: raise ValueError("Stripe subscription contains duplicate licensing SKUs") seen_skus.add(normalized_sku) quantity = item.get("quantity", 1) if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity < 1: raise ValueError("Stripe subscription item quantity must be a positive integer") period_start = _utc_from_unix(item.get("current_period_start"), "current_period_start") period_end = _utc_from_unix(item.get("current_period_end"), "current_period_end") if period_end <= period_start: raise ValueError("Stripe subscription item period is invalid") starts.append(period_start) ends.append(period_end) normalized_items.append( { "providerItemId": provider_item_id, "priceId": price_id, "skuId": normalized_sku, "quantity": quantity, "periodStart": period_start.isoformat(), "periodEnd": period_end.isoformat(), } ) event_created = _utc_from_unix(event.get("created"), "event.created") return { "provider": STRIPE_PROVIDER_ID, "providerEventId": event_id, "providerEventType": event_type, "providerEventCreatedAt": event_created.isoformat(), "subscriptionId": subscription_id, "customerId": customer_id, "licenseId": license_id, "status": status, "cancelAtPeriodEnd": bool(subscription.get("cancel_at_period_end", False)), "currentPeriodStart": max(starts).isoformat(), "currentPeriodEnd": min(ends).isoformat(), "items": normalized_items, } def normalize_stripe_billing_incident_event( event: Mapping[str, Any], ) -> dict[str, Any]: if not isinstance(event, Mapping): raise TypeError("event must be a mapping") event_id = str(event.get("id") or "").strip() event_type = str(event.get("type") or "").strip() if not event_id or len(event_id) > 256: raise ValueError("Stripe event ID is empty or too long") if event_type not in STRIPE_BILLING_INCIDENT_EVENT_TYPES: raise ValueError("Stripe billing incident event type is not supported") data = event.get("data") if not isinstance(data, Mapping) or not isinstance(data.get("object"), Mapping): raise ValueError("Stripe event data.object is missing") incident = data["object"] incident_id = str(incident.get("id") or "").strip() if not incident_id or len(incident_id) > 256: raise ValueError("Stripe billing incident ID is empty or too long") charge_reference = ( incident_id if event_type == "charge.refunded" else incident.get("charge") ) if isinstance(charge_reference, Mapping): charge_reference = charge_reference.get("id") charge_id = str(charge_reference or "").strip() if not charge_id or len(charge_id) > 256: raise ValueError("Stripe billing incident charge ID is empty or too long") incident_status = str(incident.get("status") or "").strip() if len(incident_status) > 64: raise ValueError("Stripe billing incident status is too long") event_created = _utc_from_unix(event.get("created"), "event.created") return { "provider": STRIPE_PROVIDER_ID, "providerEventId": event_id, "providerEventType": event_type, "providerEventCreatedAt": event_created.isoformat(), "incidentId": incident_id, "incidentStatus": incident_status, "chargeId": charge_id, } def enqueue_stripe_subscription_event( session: Session, normalized_event: Mapping[str, Any], raw_payload: bytes, *, now: datetime, ) -> bool: return _enqueue_stripe_provider_event( session, normalized_event, raw_payload, now=now, outbox_event_type=SUBSCRIPTION_OUTBOX_EVENT_TYPE, aggregate_type="subscription", aggregate_id_field="subscriptionId", ) def enqueue_stripe_billing_incident_event( session: Session, normalized_event: Mapping[str, Any], raw_payload: bytes, *, now: datetime, ) -> bool: return _enqueue_stripe_provider_event( session, normalized_event, raw_payload, now=now, outbox_event_type=BILLING_INCIDENT_OUTBOX_EVENT_TYPE, aggregate_type="billing_incident", aggregate_id_field="incidentId", ) def _enqueue_stripe_provider_event( session: Session, normalized_event: Mapping[str, Any], raw_payload: bytes, *, now: datetime, outbox_event_type: str, aggregate_type: str, aggregate_id_field: str, ) -> bool: if not isinstance(session, Session): raise TypeError("session must be Session") if not isinstance(normalized_event, Mapping): raise TypeError("normalized_event must be a mapping") if not isinstance(raw_payload, bytes) or not raw_payload: raise ValueError("raw_payload must be non-empty bytes") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") event_id = str(normalized_event.get("providerEventId") or "").strip() if not event_id or len(event_id) > 256: raise ValueError("normalized provider event ID is empty or too long") normalized_outbox_type = validate_identifier( outbox_event_type, "outbox_event_type", ) normalized_aggregate_type = validate_identifier( aggregate_type, "aggregate_type", ) if not isinstance(aggregate_id_field, str) or not aggregate_id_field: raise ValueError("aggregate_id_field must be a non-empty string") aggregate_id = str(normalized_event.get(aggregate_id_field) or "").strip() if not aggregate_id or len(aggregate_id) > 256: raise ValueError("normalized provider aggregate ID is empty or too long") existing = session.scalar( select(WebhookEvent).where( WebhookEvent.provider == STRIPE_PROVIDER_ID, WebhookEvent.provider_event_id == event_id, ) ) if existing is not None: return False session.add( WebhookEvent( id=new_identifier("webhook"), provider=STRIPE_PROVIDER_ID, provider_event_id=event_id, provider_object_id=aggregate_id, signature_valid=True, status="queued", payload_digest=hashlib.sha256(raw_payload).digest(), processed_at=None, error_code=None, ) ) session.add( OutboxEvent( id=new_identifier("outbox"), event_type=normalized_outbox_type, aggregate_type=normalized_aggregate_type, aggregate_id=aggregate_id, payload_json=dict(normalized_event), status="pending", available_at=now.astimezone(timezone.utc), attempts=0, processed_at=None, ) ) return True def process_subscription_outbox( session: Session, catalog: LicensingCatalog, *, now: datetime, maximum_events: int = 100, billing_incident_provider: Optional[BillingIncidentProvider] = None, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be LicensingCatalog") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if isinstance(maximum_events, bool) or not isinstance(maximum_events, int): raise TypeError("maximum_events must be an integer") if maximum_events < 1 or maximum_events > 1000: raise ValueError("maximum_events must be between one and 1000") if billing_incident_provider is not None and not callable( getattr(billing_incident_provider, "resolve_billing_incident", None) ): raise TypeError("billing_incident_provider must expose resolve_billing_incident") processed = 0 with session.begin(): events = session.scalars( select(OutboxEvent) .where( OutboxEvent.event_type.in_( ( SUBSCRIPTION_OUTBOX_EVENT_TYPE, BILLING_INCIDENT_OUTBOX_EVENT_TYPE, ) ), OutboxEvent.status == "pending", OutboxEvent.available_at <= now, ) .order_by(OutboxEvent.created_at, OutboxEvent.id) .limit(maximum_events) .with_for_update(skip_locked=True) ).all() for outbox in events: outbox.attempts += 1 webhook = session.scalar( select(WebhookEvent).where( WebhookEvent.provider == STRIPE_PROVIDER_ID, WebhookEvent.provider_event_id == outbox.payload_json["providerEventId"], ) ) try: with session.begin_nested(): if outbox.event_type == SUBSCRIPTION_OUTBOX_EVENT_TYPE: _project_subscription_event( session, catalog, outbox.payload_json, now, ) else: _project_billing_incident( session, catalog, outbox.payload_json, billing_incident_provider, now, ) except Exception as exc: outbox.status = "failed" outbox.processed_at = now outbox.last_error_code = exc.__class__.__name__.lower()[:64] if webhook is not None: webhook.status = "failed" webhook.processed_at = now webhook.error_code = exc.__class__.__name__.lower()[:64] continue outbox.status = "processed" outbox.processed_at = now outbox.last_error_code = None if webhook is not None: webhook.status = "processed" webhook.processed_at = now processed += 1 return processed def requeue_failed_subscription_outbox( session: Session, *, now: datetime, maximum_events: int = 100, maximum_attempts: int = 10, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") for field_name, value, maximum in ( ("maximum_events", maximum_events, 1000), ("maximum_attempts", maximum_attempts, MAXIMUM_OUTBOX_RETRY_ATTEMPTS), ): if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{field_name} must be an integer") if value < 1 or value > maximum: raise ValueError(f"{field_name} must be between one and {maximum}") requeued = 0 with session.begin(): events = session.scalars( select(OutboxEvent) .where( OutboxEvent.event_type.in_( ( SUBSCRIPTION_OUTBOX_EVENT_TYPE, BILLING_INCIDENT_OUTBOX_EVENT_TYPE, ) ), OutboxEvent.status == "failed", OutboxEvent.attempts < maximum_attempts, ) .order_by(OutboxEvent.created_at, OutboxEvent.id) .limit(maximum_events) .with_for_update(skip_locked=True) ).all() for outbox in events: outbox.status = "pending" outbox.available_at = now.astimezone(timezone.utc) outbox.processed_at = None outbox.last_error_code = None webhook = session.scalar( select(WebhookEvent).where( WebhookEvent.provider == STRIPE_PROVIDER_ID, WebhookEvent.provider_event_id == outbox.payload_json["providerEventId"], ) ) if webhook is not None: webhook.status = "queued" webhook.processed_at = None webhook.error_code = None requeued += 1 return requeued def _project_subscription_event( session: Session, catalog: LicensingCatalog, raw: Mapping[str, Any], now: datetime, ) -> None: license_id = validate_identifier(raw.get("licenseId"), "license_id") license_row = session.scalar( select(License).where(License.id == license_id).with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "subscription references a missing or inactive license", status_code=404, ) provider_subscription_id = str(raw.get("subscriptionId") or "").strip() provider_customer_id = str(raw.get("customerId") or "").strip() if not provider_customer_id or len(provider_customer_id) > 256: raise ValueError("subscription projection customer ID is invalid") provider_event_created_at = datetime.fromisoformat( str(raw.get("providerEventCreatedAt")).replace("Z", "+00:00") ).astimezone(timezone.utc) period_start = datetime.fromisoformat( str(raw.get("currentPeriodStart")).replace("Z", "+00:00") ).astimezone(timezone.utc) period_end = datetime.fromisoformat( str(raw.get("currentPeriodEnd")).replace("Z", "+00:00") ).astimezone(timezone.utc) if period_end <= period_start: raise ValueError("subscription projection period is invalid") status = str(raw.get("status") or "").strip() if status not in SUBSCRIPTION_STATUSES: raise ValueError("subscription projection status is invalid") subscription = session.scalar( select(Subscription) .where( Subscription.provider == STRIPE_PROVIDER_ID, Subscription.provider_subscription_id == provider_subscription_id, ) .with_for_update() ) if subscription is not None and subscription.license_id != license_row.id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "subscription cannot move between licenses", status_code=403, ) if ( subscription is not None and subscription.provider_customer_id is not None and subscription.provider_customer_id != provider_customer_id ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "subscription cannot move between provider customers", status_code=403, ) if ( subscription is not None and subscription.provider_event_created_at is not None and _utc(subscription.provider_event_created_at) >= provider_event_created_at ): return previous_status = None if subscription is None else subscription.status previous_cancel_at_period_end = ( False if subscription is None else subscription.cancel_at_period_end ) items = raw.get("items") if not isinstance(items, list) or not items: raise ValueError("subscription projection items are missing") definitions = [] for item in items: if not isinstance(item, Mapping): raise ValueError("subscription projection item must be an object") sku = catalog.require_sku(item.get("skuId"), require_active=False) if sku.kind is not SkuKind.SUBSCRIPTION: raise ValueError("subscription price maps to a non-subscription SKU") quantity = item.get("quantity") if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity < 1: raise ValueError("subscription projection quantity is invalid") definitions.append((item, sku, quantity)) policy = catalog.device_policies[definitions[0][1].device_policy_id] grace_ends_at = ( period_end + timedelta(days=policy.subscription_payment_grace_days) if status == "past_due" else None ) if subscription is None: subscription = Subscription( id=new_identifier("subscription"), license_id=license_row.id, provider=STRIPE_PROVIDER_ID, provider_subscription_id=provider_subscription_id, provider_customer_id=provider_customer_id, status=status, current_period_start=period_start, current_period_end=period_end, cancel_at_period_end=bool(raw.get("cancelAtPeriodEnd", False)), grace_ends_at=grace_ends_at, provider_event_created_at=provider_event_created_at, ) session.add(subscription) session.flush() else: subscription.provider_customer_id = provider_customer_id subscription.status = status subscription.current_period_start = period_start subscription.current_period_end = period_end subscription.cancel_at_period_end = bool(raw.get("cancelAtPeriodEnd", False)) subscription.grace_ends_at = grace_ends_at subscription.provider_event_created_at = provider_event_created_at existing_items = { value.sku_id: value for value in session.scalars( select(SubscriptionItem).where( SubscriptionItem.subscription_id == subscription.id ) ) } projected_skus = set() for item, sku, quantity in definitions: projected_skus.add(sku.sku_id) subscription_item = existing_items.get(sku.sku_id) if subscription_item is None: subscription_item = SubscriptionItem( id=new_identifier("subscription_item"), subscription_id=subscription.id, sku_id=sku.sku_id, quantity=quantity, provider_item_id=str(item.get("providerItemId") or "") or None, ) session.add(subscription_item) else: subscription_item.quantity = quantity subscription_item.provider_item_id = ( str(item.get("providerItemId") or "") or None ) source_reference = f"{STRIPE_PROVIDER_ID}:{provider_subscription_id}:{sku.sku_id}" grant = session.scalar( select(Grant).where( Grant.source_type == "subscription", Grant.source_reference == source_reference, ) ) grant_status = _subscription_grant_status( status, sku.active, subscription.billing_hold_status, ) grant_end = grace_ends_at if grace_ends_at is not None else period_end grant_metadata = _subscription_grant_metadata( status, subscription.billing_hold_status, ) if grant is None: grant = Grant( id=new_identifier("grant"), license_id=license_row.id, sku_id=sku.sku_id, source_type="subscription", source_reference=source_reference, catalog_revision=catalog.revision, starts_at=period_start, ends_at=grant_end, status=grant_status, metadata_json=grant_metadata, ) session.add(grant) else: grant.starts_at = period_start grant.ends_at = grant_end grant.status = grant_status grant.metadata_json = grant_metadata for removed_sku, removed_item in existing_items.items(): if removed_sku in projected_skus: continue session.delete(removed_item) source_reference = f"{STRIPE_PROVIDER_ID}:{provider_subscription_id}:{removed_sku}" removed_grant = session.scalar( select(Grant).where( Grant.source_type == "subscription", Grant.source_reference == source_reference, ) ) if removed_grant is not None: removed_grant.status = "revoked" removed_grant.ends_at = max( now, _utc(removed_grant.starts_at) + timedelta(seconds=1), ) for notification_kind in subscription_transition_notification_kinds( previous_status, subscription.status, previous_cancel_at_period_end, subscription.cancel_at_period_end, ): if notification_kind is CustomerNotificationKind.SUBSCRIPTION_STARTED and not any( sku.active for _item, sku, _quantity in definitions ): continue if ( notification_kind is CustomerNotificationKind.PAYMENT_RECOVERED and subscription.billing_hold_status is not None ): continue enqueue_subscription_notification( session, subscription.id, notification_kind, str(raw.get("providerEventId") or ""), now=provider_event_created_at, ) def _project_billing_incident( session: Session, catalog: LicensingCatalog, raw: Mapping[str, Any], provider: Optional[BillingIncidentProvider], now: datetime, ) -> None: if not isinstance(session, Session): raise TypeError("session must be Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be LicensingCatalog") if not isinstance(raw, Mapping): raise TypeError("raw must be a mapping") _utc(now) if provider is None or not callable(getattr(provider, "resolve_billing_incident", None)): raise ValueError("billing incident projection requires a provider API adapter") resolution = provider.resolve_billing_incident(raw) if not isinstance(resolution, ProviderBillingIncidentResolution): raise TypeError("provider returned an invalid billing incident resolution") incident_id = str(raw.get("incidentId") or "").strip() if resolution.provider_incident_id != incident_id: raise ValueError("provider resolved a different billing incident") if resolution.subscription_id is None: return subscription = session.scalar( select(Subscription) .where( Subscription.provider == STRIPE_PROVIDER_ID, Subscription.provider_subscription_id == resolution.subscription_id, ) .with_for_update() ) if subscription is None or not resolution.apply_hold: return event_created_at = datetime.fromisoformat( str(raw.get("providerEventCreatedAt")).replace("Z", "+00:00") ).astimezone(timezone.utc) if ( subscription.billing_hold_event_created_at is not None and _utc(subscription.billing_hold_event_created_at) > event_created_at ): return if ( subscription.billing_hold_status == "refunded" and resolution.hold_status != "refunded" ): return previous_hold_status = subscription.billing_hold_status subscription.billing_hold_status = resolution.hold_status subscription.billing_hold_provider_id = resolution.provider_incident_id subscription.billing_hold_event_created_at = event_created_at items = session.scalars( select(SubscriptionItem).where( SubscriptionItem.subscription_id == subscription.id ) ).all() for item in items: sku = catalog.require_sku(item.sku_id, require_active=False) source_reference = ( f"{STRIPE_PROVIDER_ID}:{subscription.provider_subscription_id}:{sku.sku_id}" ) grant = session.scalar( select(Grant).where( Grant.source_type == "subscription", Grant.source_reference == source_reference, ) ) if grant is None: continue grant.status = _subscription_grant_status( subscription.status, sku.active, subscription.billing_hold_status, ) grant.metadata_json = _subscription_grant_metadata( subscription.status, subscription.billing_hold_status, ) provider_event_id = validate_identifier( raw.get("providerEventId"), "provider_event_id", ) notification_kind = billing_hold_notification_kind( previous_hold_status, subscription.billing_hold_status, ) if notification_kind is not None: session.add( AuditEvent( id=new_identifier("audit"), actor_type="system", actor_id=None, action="subscription.billing_hold_changed", target_type="subscription", target_id=subscription.id, reason=None, correlation_id=provider_event_id, source_address_digest=None, metadata_json={ "billingHold": subscription.billing_hold_status, "providerEventType": str(raw.get("providerEventType") or ""), }, ) ) enqueue_subscription_notification( session, subscription.id, notification_kind, provider_event_id, now=event_created_at, ) def _subscription_grant_status( subscription_status: str, sku_active: bool, billing_hold_status: Optional[str], ) -> str: if subscription_status not in SUBSCRIPTION_STATUSES: raise ValueError("subscription_status is invalid") if not isinstance(sku_active, bool): raise TypeError("sku_active must be a Boolean") if billing_hold_status is not None and billing_hold_status not in ( "dispute_open", "dispute_lost", "refunded", ): raise ValueError("billing_hold_status is invalid") if billing_hold_status is not None: return "suspended" if subscription_status in ACTIVE_SUBSCRIPTION_STATUSES: return "active" if sku_active else "pending" return "suspended" def _subscription_grant_metadata( subscription_status: str, billing_hold_status: Optional[str], ) -> dict[str, object]: if subscription_status not in SUBSCRIPTION_STATUSES: raise ValueError("subscription_status is invalid") metadata: dict[str, object] = {"subscriptionState": subscription_status} if billing_hold_status is not None: metadata["billingHold"] = billing_hold_status return metadata __all__ = [ "STRIPE_BILLING_INCIDENT_EVENT_TYPES", "STRIPE_SUBSCRIPTION_EVENT_TYPES", "enqueue_stripe_billing_incident_event", "enqueue_stripe_subscription_event", "normalize_stripe_billing_incident_event", "normalize_stripe_subscription_event", "process_subscription_outbox", "requeue_failed_subscription_outbox", ]