💳 Stripe Utilities

Unified payment processing for multi-tenant SaaS: subscriptions, one-time payments, webhooks, and access control.

🎯 Overview

Category	Functions	Purpose
⚙️ Configuration	StripeConfig, StripeConfig.from_env()	Configure Stripe with env vars or explicit values
💳 Service	StripeService	Unified API for checkouts, subscriptions, webhooks
🛒 Checkout	create_subscription_checkout, create_one_time_checkout	Generate Stripe checkout sessions
🔄 Subscriptions	get_subscription, cancel_subscription, change_plan	Manage active subscriptions
🪝 Webhooks	verify_signature, handle_event	Process Stripe webhook events
🔐 Access Control	get_active_subscription, has_active_subscription, require_active_subscription	Gate features by payment status
📊 Feature Gating	check_feature_access	Control features by plan tier
🛤️ Route Helpers	create_webhook_route, create_checkout_route, create_portal_route	FastHTML route factories

---

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    Payment Flow                                  │
├─────────────────────────────────────────────────────────────────┤
│  1. User clicks "Subscribe" or "Buy"                           │
│  2. StripeService.create_*_checkout() → Stripe Session          │
│  3. User completes payment on Stripe                            │
│  4. Stripe sends webhook → handle_event()                       │
│  5. Subscription saved to core_subscriptions                    │
│  6. Access control checks subscription status                   │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Access Control                               │
├─────────────────────────────────────────────────────────────────┤
│  require_active_subscription(tenant_id)                         │
│    ├─ status = 'active' → ✅ Access granted                    │
│    ├─ status = 'trialing' → ✅ Access granted                  │
│    ├─ status = 'past_due' + within grace → ⚠️ Access granted   │
│    └─ Otherwise → 🚫 402 Payment Required                       │
└─────────────────────────────────────────────────────────────────┘

---

🌍 Environment Variables

Variable	Required	Description
STRIPE_SECRET_KEY	✅	Stripe secret API key (sk_live_* or sk_test_*)
STRIPE_WEBHOOK_SECRET	✅	Webhook endpoint signing secret (whsec_*)
STRIPE_MONTHLY_PRICE_ID	⚠️	Pre-created monthly price ID (price_*)
STRIPE_YEARLY_PRICE_ID	⚠️	Pre-created yearly price ID (price_*)
STRIPE_BASE_URL	⚠️	Application base URL for callbacks

⚠️ Price IDs are required for subscription checkouts. Create them in Stripe Dashboard first.

---

from nbdev.showdoc import show_doc

⚙️ Configuration

Configure Stripe integration with sensible defaults and environment variable support.

---

source

StripeConfig

def StripeConfig(
    secret_key:str, 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', success_path:str='/payment-success',
    cancel_path:str='/settings/payment', allow_promotions:bool=True, is_development:bool=False,
    feature_tiers:Dict=<factory>
)->None:

Configuration for Stripe integration.

Supports both subscription and one-time payment modes with configurable trial periods, grace periods, and callback URLs.

Attributes: secret_key: Stripe secret API key webhook_secret: Webhook signing secret for signature verification monthly_price_id: Pre-created Stripe price ID for monthly subscriptions yearly_price_id: Pre-created Stripe price ID for yearly subscriptions trial_days: Number of days for free trial (default: 30) grace_period_days: Days to allow access after payment failure (default: 3) base_url: Application base URL for redirect callbacks success_path: Path for successful payment redirect cancel_path: Path for canceled payment redirect allow_promotions: Enable Stripe promotion codes in checkout is_development: Enable development mode (skip signature verification)

Example: >>> config = StripeConfig.from_env() >>> service = StripeService(config)

---

source

StripeConfig.from_env

def from_env(
    
)->StripeConfig:

Create StripeConfig from environment variables.

Reads STRIPE_* environment variables with sensible defaults.

Returns: Configured StripeConfig instance

Raises: ValueError: If STRIPE_SECRET_KEY is not set

Example: >>> # Set environment variables first >>> os.environ[‘STRIPE_SECRET_KEY’] = ‘sk_test_…’ >>> config = StripeConfig.from_env()

---

💳 Stripe Service

Unified service class for all Stripe operations: checkouts, subscriptions, and webhooks.

---

source

StripeService

def StripeService(
    config:StripeConfig, host_db:HostDatabase=None
):

Unified Stripe service for payments, subscriptions, and webhooks.

Consolidates checkout creation, subscription management, and webhook handling into a single service with consistent patterns.

Attributes: config: StripeConfig instance api: FastStripe API wrapper host_db: HostDatabase for subscription persistence

Example: >>> config = StripeConfig.from_env() >>> service = StripeService(config) >>> checkout = service.create_subscription_checkout( … plan_type=‘monthly’, … tenant_id=‘tnt_123’, … user_email=‘user@example.com’ … ) >>> print(checkout.url) # Redirect user here

---

source

StripeService.create_subscription_checkout

def create_subscription_checkout(
    plan_type:str, tenant_id:str, user_email:str, metadata:Dict=None
)->Any:

Create a subscription checkout session with trial.

Args: plan_type: ‘monthly’ or ‘yearly’ tenant_id: Tenant ID to associate subscription with user_email: Customer email for Stripe metadata: Additional metadata to store with subscription

Returns: Stripe Checkout Session with ‘url’ and ‘id’ attributes

Raises: ValueError: If plan_type is invalid or price ID not configured

Example: >>> checkout = service.create_subscription_checkout( … plan_type=‘monthly’, … tenant_id=‘tnt_abc123’, … user_email=‘user@example.com’ … ) >>> return RedirectResponse(checkout.url)

---

source

StripeService.create_one_time_checkout

def create_one_time_checkout(
    amount_cents:int, product_name:str, tenant_id:str, user_email:str, currency:str='usd', metadata:Dict=None
)->Any:

Create a one-time payment checkout session.

Args: amount_cents: Payment amount in cents (e.g., 1999 for $19.99) product_name: Name displayed on checkout tenant_id: Tenant ID for record keeping user_email: Customer email for Stripe currency: ISO currency code (default: ‘usd’) metadata: Additional metadata to store

Returns: Stripe Checkout Session with ‘url’ and ‘id’ attributes

Example: >>> checkout = service.create_one_time_checkout( … amount_cents=4999, … product_name=‘Premium Report’, … tenant_id=‘tnt_abc123’, … user_email=‘user@example.com’ … ) >>> return RedirectResponse(checkout.url)

---

source

StripeService.create_customer_portal_session

def create_customer_portal_session(
    customer_id:str, return_url:str=None
)->Any:

Create a Stripe Customer Portal session for self-service billing.

Args: customer_id: Stripe customer ID (cus_*) return_url: URL to redirect after portal session (default: base_url)

Returns: Portal session with ‘url’ attribute

Example: >>> portal = service.create_customer_portal_session(‘cus_123’) >>> return RedirectResponse(portal.url)

---

source

StripeService.cancel_subscription

def cancel_subscription(
    subscription_id:str, at_period_end:bool=True
)->Any:

Cancel a subscription.

Args: subscription_id: Stripe subscription ID at_period_end: If True, cancel at end of billing period (default) If False, cancel immediately

Returns: Updated Stripe Subscription object

---

source

StripeService.handle_event

def handle_event(
    event:Dict
)->Dict:

Route webhook event to appropriate handler.

Handles both subscription and one-time payment events.

Args: event: Parsed Stripe event dict

Returns: Dict with ‘status’ (‘success’, ‘warning’, ‘error’, ‘ignored’) and ‘message’ describing the result

---

💰 Pricing Plans

Database-backed pricing tiers for multi-tier subscription management. Store your Stripe price IDs in the database for admin-configurable pricing without code changes.

---

source

get_pricing_plans

def get_pricing_plans(
    host_db:HostDatabase=None, active_only:bool=True
)->List:

Get all pricing plans from the database.

Retrieves pricing plan configurations stored in the core_pricing_plans table. Plans define available subscription tiers with their Stripe price IDs, amounts, features, and display settings.

Args: host_db: Optional HostDatabase instance. Uses from_env() if not provided. active_only: If True (default), only returns plans where is_active=True. Set to False to include inactive/archived plans.

Returns: List of PricingPlan objects sorted by sort_order, then by tier_level. Empty list if no plans are configured.

Example: >>> # Get all active plans for pricing page >>> plans = get_pricing_plans() >>> for plan in plans: … print(f”{plan.name}: ${plan.amount_monthly/100}/mo”) Basic Plan: $7.99/mo Pro Plan: $19.99/mo Enterprise Plan: $49.99/mo

>>> # Get specific plan for checkout
>>> plans = get_pricing_plans()
>>> pro_plan = next((p for p in plans if p.id == 'pro'), None)
>>> if pro_plan:
...     price_id = pro_plan.stripe_price_monthly

>>> # Setting up plans (typically in admin or migration)
>>> from fh_saas.db_host import HostDatabase, PricingPlan, gen_id, timestamp
>>> host_db = HostDatabase.from_env()
>>> 
>>> plans_data = [
...     {
...         'id': 'basic',
...         'name': 'Basic Plan',
...         'description': 'Essential features for individuals',
...         'stripe_price_monthly': 'price_basic_monthly_xxx',
...         'stripe_price_yearly': 'price_basic_yearly_xxx',
...         'amount_monthly': 799,   # $7.99
...         'amount_yearly': 7990,   # $79.90 (save ~17%)
...         'tier_level': 1,
...         'features': '["basic_reports", "email_support"]',
...         'sort_order': 1,
...     },
...     {
...         'id': 'pro',
...         'name': 'Pro Plan', 
...         'description': 'Advanced features for teams',
...         'stripe_price_monthly': 'price_pro_monthly_xxx',
...         'stripe_price_yearly': 'price_pro_yearly_xxx',
...         'amount_monthly': 1999,  # $19.99
...         'amount_yearly': 19990,  # $199.90
...         'tier_level': 2,
...         'features': '["basic_reports", "advanced_analytics", "api_access", "priority_support"]',
...         'sort_order': 2,
...     },
...     {
...         'id': 'enterprise',
...         'name': 'Enterprise Plan',
...         'description': 'Full platform access with custom solutions',
...         'stripe_price_monthly': 'price_ent_monthly_xxx',
...         'stripe_price_yearly': 'price_ent_yearly_xxx',
...         'amount_monthly': 4999,  # $49.99
...         'amount_yearly': 49990,  # $499.90
...         'tier_level': 3,
...         'features': '["all_features", "dedicated_support", "custom_integrations", "sla"]',
...         'sort_order': 3,
...     },
... ]
>>> 
>>> for plan_data in plans_data:
...     plan = PricingPlan(**plan_data, created_at=timestamp())
...     host_db.pricing_plans.insert(plan)
>>> host_db.commit()

Note: - Create your Stripe Products and Prices in the Stripe Dashboard first - Copy the price IDs (price_xxx) into your database records - The tier_level field is used by check_feature_access() for feature gating - The features field should be a JSON array of feature keys

---

source

get_pricing_plan

def get_pricing_plan(
    plan_id:str, host_db:HostDatabase=None
)->Optional:

Get a specific pricing plan by ID.

Args: plan_id: The plan identifier (e.g., ‘basic’, ‘pro’, ‘enterprise’) host_db: Optional HostDatabase instance

Returns: PricingPlan object if found, None otherwise

Example: >>> plan = get_pricing_plan(‘pro’) >>> if plan: … checkout = service.create_subscription_checkout( … price_id=plan.stripe_price_monthly, … … … )

---

🔐 Access Control

Functions to gate features based on subscription status with grace period support.

---

source

get_active_subscription

def get_active_subscription(
    tenant_id:str, host_db:HostDatabase=None, grace_period_days:int=3
)->Optional:

Get active subscription for a tenant with grace period support.

Returns subscription if: - Status is ‘active’ or ‘trialing’ - Status is ‘past_due’ but within grace period from current_period_end

Args: tenant_id: Tenant ID to check host_db: Optional HostDatabase, uses from_env() if not provided grace_period_days: Days to allow access after payment failure (default: 3)

Returns: Active Subscription object, or None if no valid subscription

Example: >>> sub = get_active_subscription(‘tnt_123’) >>> if sub: … print(f”Active until {sub.current_period_end}“)

---

source

has_active_subscription

def has_active_subscription(
    tenant_id:str, host_db:HostDatabase=None, grace_period_days:int=3
)->bool:

Check if tenant has an active subscription.

Args: tenant_id: Tenant ID to check host_db: Optional HostDatabase grace_period_days: Days to allow access after payment failure

Returns: True if tenant has valid subscription, False otherwise

Example: >>> if has_active_subscription(‘tnt_123’): … show_premium_features()

---

source

require_active_subscription

def require_active_subscription(
    tenant_id:str, host_db:HostDatabase=None, grace_period_days:int=3, redirect_url:str=None
)->Optional:

Require active subscription, returning 402 if not found.

Use in route handlers to gate premium features.

Args: tenant_id: Tenant ID to check host_db: Optional HostDatabase grace_period_days: Days to allow access after payment failure redirect_url: Optional URL to redirect instead of 402

Returns: None if subscription is valid (allow access) Response(402) or RedirectResponse if subscription required

Example: >>> @app.get(‘/premium-feature’) >>> def premium_feature(request): … error = require_active_subscription(request.state.tenant_id) … if error: … return error … return render_premium_content()

---

source

get_subscription_status

def get_subscription_status(
    tenant_id:str, host_db:HostDatabase=None, grace_period_days:int=3
)->Dict:

Get detailed subscription status for UI display.

Args: tenant_id: Tenant ID to check host_db: Optional HostDatabase grace_period_days: Days for grace period calculation

Returns: Dict with subscription details for UI: - has_subscription: bool - status: str (‘active’, ‘trialing’, ‘past_due’, ‘canceled’, ‘none’) - plan_tier: str or None - is_trial: bool - trial_ends_at: str (ISO) or None - current_period_end: str (ISO) or None - days_remaining: int or None - in_grace_period: bool - cancel_at_period_end: bool

Example: >>> status = get_subscription_status(‘tnt_123’) >>> if status[‘is_trial’]: … show_trial_banner(status[‘days_remaining’])

---

📊 Feature Gating

Control access to features based on subscription plan tier.

---

source

check_feature_access

def check_feature_access(
    tenant_id:str, feature:str, feature_tiers:Dict=None, host_db:HostDatabase=None, grace_period_days:int=3
)->bool:

Check if tenant’s plan tier allows access to a feature.

Args: tenant_id: Tenant ID to check feature: Feature name to check access for feature_tiers: Optional mapping of feature -> required tier. Defaults to {‘advanced_analytics’: ‘yearly’, …} host_db: Optional HostDatabase grace_period_days: Days for grace period

Returns: True if tenant’s plan allows the feature, False otherwise

Example: >>> if check_feature_access(‘tnt_123’, ‘advanced_analytics’): … return render_analytics() >>> else: … return render_upgrade_prompt()

---

source

require_feature_access

def require_feature_access(
    tenant_id:str, feature:str, feature_tiers:Dict=None, host_db:HostDatabase=None, redirect_url:str=None
)->Optional:

Require feature access, returning 403 if not allowed.

Args: tenant_id: Tenant ID to check feature: Feature name to check feature_tiers: Optional feature -> tier mapping host_db: Optional HostDatabase redirect_url: Optional URL to redirect instead of 403

Returns: None if access allowed, Response(403) or redirect otherwise

Example: >>> @app.get(‘/analytics’) >>> def analytics(request): … error = require_feature_access( … request.state.tenant_id, … ‘advanced_analytics’, … redirect_url=‘/upgrade’ … ) … if error: … return error … return render_analytics()

---

🛤️ Route Helpers

Ready-to-use FastHTML route factories for Stripe integration.

---

source

create_webhook_route

def create_webhook_route(
    app, service:StripeService, path:str='/stripe/webhook'
):

Create a POST route handler for Stripe webhooks.

Args: app: FastHTML application instance service: Configured StripeService instance path: URL path for webhook endpoint

Returns: The registered route handler

Example: >>> from fh_saas.utils_stripe import StripeService, StripeConfig, create_webhook_route >>> config = StripeConfig.from_env() >>> service = StripeService(config) >>> create_webhook_route(app, service) >>> # Webhook now available at POST /stripe/webhook

---

source

create_subscription_checkout_route

def create_subscription_checkout_route(
    app, service:StripeService, path:str='/checkout/{plan_type}'
):

Create a GET route for subscription checkout redirect.

Expects authenticated user with tenant_id in request.state.

Args: app: FastHTML application instance service: Configured StripeService instance path: URL path pattern with {plan_type} placeholder

Returns: The registered route handler

Example: >>> create_subscription_checkout_route(app, service) >>> # Checkout available at GET /checkout/monthly or /checkout/yearly

---

source

create_one_time_checkout_route

def create_one_time_checkout_route(
    app, service:StripeService, products:Dict, path:str='/buy/{product_id}'
):

Create a GET route for one-time payment checkout.

Args: app: FastHTML application instance service: Configured StripeService instance products: Dict mapping product_id to {‘name’: str, ‘amount_cents’: int} path: URL path pattern with {product_id} placeholder

Returns: The registered route handler

Example: >>> products = { … ‘report’: {‘name’: ‘Premium Report’, ‘amount_cents’: 4999}, … ‘credits’: {‘name’: ‘100 Credits’, ‘amount_cents’: 1999}, … } >>> create_one_time_checkout_route(app, service, products) >>> # Checkout available at GET /buy/report or /buy/credits

---

source

create_portal_route

def create_portal_route(
    app, service:StripeService, path:str='/billing-portal'
):

Create a GET route for Stripe Customer Portal redirect.

Args: app: FastHTML application instance service: Configured StripeService instance path: URL path for portal endpoint

Returns: The registered route handler

Example: >>> create_portal_route(app, service) >>> # Portal available at GET /billing-portal

---

🏭 Service Factory

Singleton pattern for easy service access across the application.

---

source

reset_stripe_service

def reset_stripe_service(
    
):

Reset singleton instance (for testing only).

---

source

get_stripe_service

def get_stripe_service(
    config:StripeConfig=None
)->StripeService:

Get or create singleton StripeService instance.

Args: config: Optional StripeConfig. Uses from_env() on first call if not provided.

Returns: StripeService singleton instance

Example: >>> service = get_stripe_service() >>> checkout = service.create_subscription_checkout(…)

---

🚀 Quick Start

1. Set Environment Variables

# Required
export STRIPE_SECRET_KEY="sk_test_..."
export STRIPE_WEBHOOK_SECRET="whsec_..."

# Create prices in Stripe Dashboard, then set IDs
export STRIPE_MONTHLY_PRICE_ID="price_..."
export STRIPE_YEARLY_PRICE_ID="price_..."

# Application URL
export STRIPE_BASE_URL="https://yourapp.com"

2. Initialize Service & Routes

from fasthtml.common import FastHTML
from fh_saas.utils_stripe import (
    get_stripe_service,
    create_webhook_route,
    create_subscription_checkout_route,
    create_portal_route,
)

app = FastHTML()
service = get_stripe_service()

# Register routes
create_webhook_route(app, service)
create_subscription_checkout_route(app, service)
create_portal_route(app, service)

# Now available:
# POST /stripe/webhook       - Stripe webhook handler
# GET  /checkout/monthly     - Monthly subscription checkout
# GET  /checkout/yearly      - Yearly subscription checkout
# GET  /billing-portal       - Customer self-service portal

3. Gate Premium Features

from fh_saas.utils_stripe import require_active_subscription

@app.get('/premium')
def premium_feature(request):
    error = require_active_subscription(request.state.tenant_id)
    if error:
        return error  # 402 Payment Required
    return render_premium_content()

4. Integrate with Auth Beforeware

from fh_saas.utils_auth import create_auth_beforeware

# Optional: Add require_subscription=True to beforeware
# See utils_auth for integration examples