๐Ÿงพ Billing Orchestrator

Provider-agnostic billing contracts, lifecycle services, and one-liner route registration for multi-tenant SaaS.

source

SubscriptionStatus


def SubscriptionStatus(
    args:VAR_POSITIONAL, kwds:VAR_KEYWORD
):

Canonical subscription status values.

Every subscription in the system maps to one of these statuses. Use SubscriptionStatus.normalize() to map raw Stripe/legacy values.


source

BillingConfig


def BillingConfig(
    stripe_secret_key:str='', stripe_webhook_secret:str=None, monthly_price_id:str=None, yearly_price_id:str=None,
    trial_days:int=30, grace_period_days:int=3, base_url:str='http://localhost:5001', pricing_path:str='/pricing',
    checkout_path:str='/checkout/{plan_type}', success_path:str='/payment-success',
    cancel_path:str='/settings/payment', webhook_path:str='/stripe/webhook', portal_path:str='/billing-portal',
    pricing_mode:str='public_and_post_login_pricing', allow_promotions:bool=True, is_development:bool=False
)->None:

Configuration for the billing orchestrator.


source

validate_billing_config


def validate_billing_config(
    config:BillingConfig
)->List[str]:

Validate billing configuration and return actionable error messages.


source

AccessDecision


def AccessDecision(
    allowed:bool, reason:SubscriptionStatus, redirect_to:Optional[str]=None
)->None:

Result of a subscription access check.

source

PlanStore


def PlanStore(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Interface for accessing pricing plan data.


source

SubscriptionStore


def SubscriptionStore(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Interface for subscription persistence and webhook idempotency.


source

UserContextProvider


def UserContextProvider(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Interface for extracting user context from HTTP requests.


source

BillingUIAdapter


def BillingUIAdapter(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Optional UI rendering callbacks for billing routes (headless-first).

Default Implementations


source

HostDBPlanStore


def HostDBPlanStore(
    host_db
):

Default PlanStore backed by HostDatabase.


source

HostDBSubscriptionStore


def HostDBSubscriptionStore(
    host_db
):

Default SubscriptionStore backed by HostDatabase.


source

RequestUserContext


def RequestUserContext(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Default UserContextProvider reading from request.state.user.

source

init_trial_on_first_login


def init_trial_on_first_login(
    tenant_id:str, user_email:str, subscription_store:SubscriptionStore, config:BillingConfig
):

Create a trial subscription if the tenant has no subscription yet and return the resolved subscription.


source

resolve_subscription_access


def resolve_subscription_access(
    subscription, config:BillingConfig, now:datetime=None
)->AccessDecision:

Determine whether a user should be allowed access based on subscription state. Pure function.


source

resolve_checkout_return


def resolve_checkout_return(
    session_id:str, tenant_id:str, subscription_store:SubscriptionStore, subscription:NoneType=None
)->SubscriptionStatus:

Determine subscription state after a checkout redirect.

source

register_billing_routes


def register_billing_routes(
    app, config:BillingConfig, plan_store:PlanStore=None, subscription_store:SubscriptionStore=None,
    user_ctx:UserContextProvider=None, stripe_service:NoneType=None, ui_adapter:BillingUIAdapter=None,
    host_db:NoneType=None
):

Wire all billing routes onto a FastHTML app in one call.


source

attach_billing_auth_hooks


def attach_billing_auth_hooks(
    app, config:BillingConfig, subscription_store:SubscriptionStore=None, user_ctx:UserContextProvider=None,
    host_db:NoneType=None, skip_paths:List[str]=None
):

Attach first-login trial init and paywall guard as app middleware.