--- title: "Stripe Cancellations in FastAPI and SQLModel" description: "Today I am working on fokais.com, trying to get to a point where I can launch by workig through stripe integrations. This is my first time using stripe, so..." date: 2023-12-09 published: true tags: - python template: til --- Today I am working on fokais.com, trying to get to a point where I can launch by workig through stripe integrations. This is my first time using stripe, so there has been quite a bit to learn, and I am probably building in more than I need to before launching, but I am learning, and not in a rush to launch. I am building the fokais backent in python primarilyt with fastapi and sqlmodel on sqlite. My billing integration is going to be all Stripe. ## Stripe Subscription Cancellations Docs Here is a link to the stripe docs for your refrence, especially if you want to see how to cancel subscriptions in other languages. They include code samples for many popular languages. Cancel subscriptions | Stripe Documentation ## User Model This is the part of the user model that includes the cancel and reactivate methods. It pretty much follows the stripe guide. ```python class UserBase(SQLModel, table=False): # type: ignore[call-arg] username: str = Field(unique=True) full_name: str email: str email_verified: bool = False disabled: bool = False signup_date: Optional[datetime] = Field(default_factory=datetime.utcnow) stripe_customer_id: Optional[str] def cancel_subscription(self): for subscription in self.active_subscriptions: stripe.Subscription.modify( subscription.id, cancel_at_period_end=True, ) self.refresh() def reactivate_subscription(self): for subscription in self.active_subscriptions: stripe.Subscription.modify( subscription.id, cancel_at_period_end=False, ) self.refresh() ``` ## Cancellations api Here is the cancellations api. I created an are you sure form that I can link to from the accounts page with a normal anchor tag. Note that I am doing a `POST` request to do the cancellation from a form. I want this to work for any user whether there is js or not. This is an operation that will change the users data, and I want to make sure that it avoids all browser and cdn caching. As a scrappy startup we are running light on infrastructure and are caching hard at the CDN to avoid excessive server hits. !!! Note I am doing a `POST` request to do the cancellation from a form. ```python @pricing_router.get("/cancel") @pricing_router.get("/cancel/") def get_cancel( request: Request, current_user: Annotated[User, Depends(get_current_user_if_logged_in)], ): return config.templates.TemplateResponse( "cancel.html", { "request": request, "prices": products.prices, "products": products.products, "current_user": current_user, }, ) @pricing_router.post("/cancel") @pricing_router.post("/cancel/") def post_cancel( request: Request, current_user: Annotated[User, Depends(get_current_user_if_logged_in)], ): current_user.cancel_subscription() return HTMLResponse('

Your Subscription has been Cancelled

') ``` ## Reactivations Reactivating accounts looks just about the same as cancelling, only flippng `True` to `False`. ```python @pricing_router.get("/reactivate") @pricing_router.get("/reactivate/") def get_reactivate( request: Request, current_user: Annotated[User, Depends(get_current_user_if_logged_in)], ): return config.templates.TemplateResponse( "reactivate.html", { "request": request, "prices": products.prices, "products": products.products, "current_user": current_user, }, ) @pricing_router.post("/reactivate") @pricing_router.post("/reactivate/") def post_reactivate( request: Request, current_user: Annotated[User, Depends(get_current_user_if_logged_in)], ): current_user.reactivate_subscription() return HTMLResponse('

Your Subscription has been reactivated

') ``` ## Full User Model This is the full user model, completely subject to change in the future, but it includes the cancel and reactivate methods. ```python class UserBase(SQLModel, table=False): # type: ignore[call-arg] username: str = Field(unique=True) full_name: str email: str email_verified: bool = False disabled: bool = False signup_date: Optional[datetime] = Field(default_factory=datetime.utcnow) stripe_customer_id: Optional[str] @property def session(self): return next(get_session()) @classmethod def get_by_id(cls, id): return next(get_session()).get(cls, id) def refresh(self): cache.set(f"active_subscriptions_{self.id}", None, 3600) cache.set(f"active_products_{self.id}", None, 3600) def get_checkout_sessions(self): return [ stripe.checkout.Session.retrieve(s.stripe_checkout_session_id) for s in self.session.exec(select(CheckoutSession).where(CheckoutSession.user_id == self.id)).all() if s.stripe_checkout_session_id is not None ] def get_active_subscriptions(self): subscriptions = [ s.subscription for s in [ stripe.checkout.Session.retrieve(s.stripe_checkout_session_id) for s in self.session.exec(select(CheckoutSession).where(CheckoutSession.user_id == self.id)).all() if s.stripe_checkout_session_id is not None ] if s.status == "complete" ] active_subscriptions = [stripe.Subscription.retrieve(subscription) for subscription in subscriptions] return active_subscriptions def has_active_subscription(self): return len(self.active_subscriptions) > 0 @property def active_subscriptions(self): active_subscriptions = cache.get(f"active_subscriptions_{self.id}") if active_subscriptions is not None: return active_subscriptions active_subscriptions = self.get_active_subscriptions() cache.set(f"active_subscriptions_{self.id}", active_subscriptions, 3600) return active_subscriptions @property def active_plans(self): subscriptions = self.active_subscriptions plans = [subscription.plan for subscription in subscriptions] return plans @property def subscription_to_plan(self): subscriptions = self.active_subscriptions plans = {subscription.id: subscription.plan.id for subscription in subscriptions} return plans @property def plan_to_subscription(self): plans = {v: k for k, v in self.subscription_to_plan.items()} return plans def get_active_products(self): plans = self.active_plans products = [stripe.Product.retrieve(plan.product) for plan in plans] return products @property def plan_to_product(self): plans = self.active_plans products = {plan.id: stripe.Product.retrieve(plan.product).id for plan in plans} return products @property def prodct_to_plan(self): plans = self.active_plans products = {stripe.Product.retrieve(plan.product).id: plan.id for plan in plans} return products @property def active_products(self): products = cache.get(f"active_products_{self.id}") if products is not None: return products products = self.get_active_products() cache.set(f"active_products_{self.id}", products, 3600) return products @property def best_active_subscription(self): subscriptions = self.active_subscriptions return subscriptions[0] @property def best_active_product(self): products = self.active_products products.sort(key=lambda p: p.metadata.get('level', 0)) return products[0] @property def best_active_subscription(self): subscription_id = self.plan_to_subscription[self.prodct_to_plan[self.best_active_product.id]] return stripe.Subscription.retrieve(subscription_id) @property def config(self): product = self.best_active_product return product.metadata def subscription_status(self): subscriptions = self.active_subscriptions() def cancel_subscription(self): for subscription in self.active_subscriptions: stripe.Subscription.modify( subscription.id, cancel_at_period_end=True, ) self.refresh() def reactivate_subscription(self): for subscription in self.active_subscriptions: stripe.Subscription.modify( subscription.id, cancel_at_period_end=False, ) self.refresh() ```