Hey @adrierre.ae not sure if his is on your radar anymore, but I was recently trying to address the same issue and couldn’t find a way using the Anvil Stripe integration, so I’ve gone full-on Stripe API. Here’s how I implemented a simple subscription workflow using Stripe API.
First I create a form, this one works for Stripe test and live mode, with two simple subscriptions, “Basic” and “Pro”. The trickiest part here is the javascript to launch the Stripe payment window.
from ._anvil_designer import PricingTemplate
from anvil import *
import anvil.server
import anvil.users
import anvil.js
class Pricing(PricingTemplate):
def __init__(self, **properties):
# Define mode
self.mode = "live"
# Set Form properties and Data Bindings.
self.init_components(**properties)
self.link_2.foreground = "gray"
self.user = anvil.users.get_user() or anvil.users.login_with_form()
# Inject Stripe.js script into the document
anvil.js.window.eval("""
var script = document.createElement('script');
script.src = 'https://js.stripe.com/v3/';
document.head.appendChild(script);
""")
# Check and initialize Stripe object when script is loaded
self.wait_for_stripe()
def wait_for_stripe(self):
def init_stripe():
if anvil.js.window.get('Stripe'):
# Initialize Stripe after ensuring the library is loaded
if self.mode == 'live':
publishable_key = 'YOUR_LIVE_PUBLISHABLE_KEY'
else:
publishable_key = 'YOUR_TEST_PUBLISHABLE_KEY'
# Initialize Stripe with the correct key
anvil.js.window.stripe = anvil.js.window.Stripe(publishable_key)
else:
# Wait a moment and check again if Stripe.js is not loaded yet
anvil.js.window.setTimeout(self.wait_for_stripe, 100)
init_stripe()
# Inside Pricing class
def handle_plan_buttons(self, selected_plan_id, plan_name):
"""Generic function to handle plan button clicks."""
# Call server to check subscription status
result = anvil.server.call('user_already_subscribed', selected_plan_id)
if result == 'same_plan':
anvil.alert(f"You are already subscribed to the {plan_name} plan.")
elif result == 'different_plan':
# Update the existing subscription to the new plan
update_result = anvil.server.call('update_stripe_subscription', selected_plan_id)
if update_result['status'] == 'success':
# Show alert with the updated information
anvil.alert(
f"• Your subscription has been successfully updated to the {update_result['plan_name']} plan.\n"
f"• Your balance was increased by {update_result['balance_increment']}.\n"
f"• Your plan will be automatically renewed on {update_result['next_renewal']}.\n",
title="Subscription Update"
)
else:
anvil.alert(f"An error occurred: {update_result['message']}")
elif result == 'no_subscription':
session_id = anvil.server.call('create_stripe_checkout_session', plan=selected_plan_id)
anvil.js.window.stripe.redirectToCheckout({'sessionId': session_id}).then(
lambda result: anvil.alert(f"Error: {result.error.message}") if result.get('error') else None
)
else:
anvil.alert("An error occurred. Please try again later.")
# Function to determine the selected plan based on mode
def get_plan_id(self, plan_type):
plan_mapping = {
'test': {
'Basic': 'YOUR_TEST_BASIC_PLAN_ID',
'Pro': 'YOUR_TEST_PRO_PLAN_ID',
},
'live': {
'Basic': 'YOUR_LIVE_BASIC_PLAN_ID',
'Pro': 'YOUR_LIVE_PRO_PLAN_ID',
}
}
return plan_mapping[self.mode].get(plan_type)
def basic_plan_button_click(self, **event_args):
"""Handle Basic Plan button click."""
selected_plan_id = self.get_plan_id('Basic') # Get correct Basic plan ID
self.handle_plan_buttons(selected_plan_id, "Basic")
def pro_plan_button_click(self, **event_args):
"""Handle Pro Plan button click."""
selected_plan_id = self.get_plan_id('Pro') # Get correct Pro plan ID
self.handle_plan_buttons(selected_plan_id, "Pro")
There are a few server functions associated with this. First, check if the user is already subscribed to the chosen plan:
@anvil.server.callable
def user_already_subscribed(selected_plan_id):
user = anvil.users.get_user()
while not user:
anvil.users.login_with_form()
stripe.api_key = anvil.secrets.get_secret(f'stripe_key_{mode}')
if user['subscription_id']:
try:
subscription = stripe.Subscription.retrieve(user['subscription_id'])
if subscription.get('status') != 'active':
return 'no_subscription'
current_plan_id = subscription['items']['data'][0]['price']['id']
if current_plan_id == selected_plan_id:
return 'same_plan'
else:
return 'different_plan'
except stripe.error.InvalidRequestError as e:
print(f"Stripe error: {e}")
return 'no_subscription'
except Exception as e:
print(f"An unexpected error occurred: {str(e)}")
return 'error'
else:
return 'no_subscription'
If there is no subscription, we send the user to the Stripe checkout page, and then route them back to our app through a success/failure URL.
@anvil.server.callable
def create_stripe_checkout_session(plan):
user = anvil.users.get_user()
while not user:
anvil.users.login_with_form()
stripe.api_key = anvil.secrets.get_secret(f'stripe_key_{mode}')
try:
# Log the start of the process and the user info
print(f"Starting checkout session for user: {user}")
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price': plan, # Pass the plan ID
'quantity': 1,
}],
mode='subscription', # Indicates that the session is for a subscription
success_url='YOUR SUCCESS URL', # Redirect here on success
cancel_url='YOUR FAILURE URL', # Redirect here on cancellation
)
# Log the session creation result
print(f"Stripe session created with ID: {session['id']}")
# Save the session ID to the user's record and log the action
user['last_session_id'] = session['id']
print(f"Session ID {session['id']} successfully saved to user {user}")
# Return the session ID to the client
return session['id']
except Exception as e:
# Log the error and re-raise
print(f"Error creating Stripe checkout session: {str(e)}")
raise anvil.server.InternalError(f"Error creating Stripe checkout session: {str(e)}")
except Exception as e:
print(f"Error creating Stripe checkout session: {str(e)}")
raise anvil.server.InternalError(f"Error creating Stripe checkout session: {str(e)}")
If there is a subscription, we update it.
@anvil.server.callable
def update_stripe_subscription(new_plan_id):
user = anvil.users.get_user()
if not user or not user['subscription_id']:
return "Error: No active subscription found."
try:
stripe.api_key = anvil.secrets.get_secret(f'stripe_key_{mode}')
# Retrieve the existing subscription
subscription = stripe.Subscription.retrieve(user['subscription_id'])
subscription_item_id = subscription['items']['data'][0]['id'] # Get the subscription item ID
# Update the existing subscription to the new plan
stripe.Subscription.modify(
user['subscription_id'],
cancel_at_period_end=False,
billing_cycle_anchor='now', # Reset the billing cycle to charge immediately
proration_behavior='none',
items=[{
'id': subscription_item_id,
'price': new_plan_id, # Set to the new plan ID
}]
)
# Update the local user record with the new plan details
try:
# Access the plan name using square brackets, handle KeyError if plan ID is invalid
plan_name, balance_increment = plan_mapping[mode][new_plan_id]
except KeyError:
return {"status": "error", "message": "Invalid plan ID."}
# Update the local user record with the new plan details
user['current_plan'] = plan_name
user['next_renewal'] = datetime.fromtimestamp(subscription['current_period_end']).date()
user['balance'] = (user['balance'] or 0) + balance_increment
return {
"status": "success",
"plan_name": plan_name,
"balance_increment": balance_increment,
"next_renewal": user['next_renewal'].strftime('%Y-%m-%d')
}
except stripe.error.StripeError as e:
print(f"Stripe error: {e}")
return {"status": "error", "message": e.user_message}
except Exception as e:
print(f"An unexpected error occurred: {str(e)}")
return {"status": "error", "message": str(e)}
The final trick is to update the subscription status once the user have completed their payment on Stripe. The way I did this is to save the Stripe session_id to the User table before sending the user to Stripe. When the user gets back, I check for a session_id and update their subscription accordingly.
@anvil.server.callable
def handle_subscription_completed(session_id):
"""Handle successful subscription after checkout session is completed."""
user = anvil.users.get_user()
if not user:
raise ValueError("User is not logged in.")
try:
stripe.api_key = anvil.secrets.get_secret(f'stripe_key_{mode}')
# Retrieve the session and subscription details
session = stripe.checkout.Session.retrieve(session_id)
subscription = stripe.Subscription.retrieve(session.subscription)
# Log the retrieved session and subscription details
print(f"Stripe session: {session}")
print(f"Stripe subscription: {subscription}")
# Extract the plan from the subscription
price_id = subscription['items']['data'][0]['price']['id']
plan_name, balance_increment = plan_mapping[mode][price_id]
# Log the price_id, plan_name, and balance_increment
print(f"Price ID: {price_id}, Plan Name: {plan_name}, Balance Increment: {balance_increment}")
# Check if the balance increment is valid
if balance_increment == 0:
print("Balance increment is 0. There may be an issue with plan_mapping.")
return "Error: Invalid plan mapping."
# Update the user with subscription and balance information
user['stripe_customer'] = subscription.customer
user['subscription_id'] = subscription.id
user['date_subscribed'] = datetime.now().date()
user['next_renewal'] = datetime.fromtimestamp(subscription['current_period_end']).date()
# Log the current balance before updating
print(f"Current user balance: {user['balance']}")
user['balance'] = (user['balance'] or 0) + balance_increment
user['current_plan'] = plan_name
user['status'] = subscription['status']
# Log the updated user balance
print(f"Updated user balance: {user['balance']}")
return f"Subscription updated to {plan_name}."
except stripe.error.InvalidRequestError as e:
print(f"Stripe error: {e}")
return f"Error handling subscription: {e.user_message}"
except Exception as e:
print(f"An unexpected error occurred: {str(e)}")
return f"Error: {str(e)}"
This is a ton of code but gets the job done. Hope it’s useful, but let me know if I can improve or simplify!