Python's ecosystem — Stripe SDK, SQLAlchemy, and FastAPI/Django — provides a productive foundation for building subscription billing systems. This guide covers the full implementation: data models, Stripe integration, webhook handling, usage metering, and proration calculations, all with type hints and production-ready patterns.
Data Models with SQLAlchemy
python
1from sqlalchemy import Column, String, Integer, BigInteger, Boolean, DateTime, JSON, Enum, ForeignKey
2from sqlalchemy.orm import declarative_base, relationship
3from sqlalchemy.sql import func
4import enum
5from datetime import datetime
6
7Base = declarative_base()
8
9class SubscriptionStatus(enum.Enum):
10 ACTIVE = "active"
11 TRIALING = "trialing"
12 PAST_DUE = "past_due"
13 CANCELLED = "cancelled"
14 PAUSED = "paused"
15
16class BillingInterval(enum.Enum):
17 MONTH = "month"
18 YEAR = "year"
19
20class Plan(Base):
21 __tablename__ = "plans"
22
23 id = Column(String, primary_key=True)
24 name = Column(String, nullable=False)
25 slug = Column(String, unique=True, nullable=False)
26 active = Column(Boolean, default=True)
27 billing_interval = Column(Enum(BillingInterval), nullable=False)
28 base_price_cents = Column(BigInteger, nullable=False)
29 price_per_seat_cents = Column(BigInteger, default=0)
30 included_seats = Column(Integer, default=1)
31 features = Column(JSON, default=list)
32 max_projects = Column(Integer, nullable=True)
33 max_storage_mb = Column(BigInteger, default=5120)
34 max_api_calls = Column(Integer, nullable=True)
35 stripe_price_id = Column(String, nullable=False)
36 created_at = Column(DateTime, server_default=func.now())
37
38class Subscription(Base):
39 __tablename__ = "subscriptions"
40
41 id = Column(String, primary_key=True)
42 customer_id = Column(String, nullable=False, index=True)
43 plan_id = Column(String, ForeignKey("plans.id"), nullable=False)
44 stripe_subscription_id = Column(String, unique=True, nullable=False)
45 stripe_customer_id = Column(String, nullable=False)
46 status = Column(Enum(SubscriptionStatus), default=SubscriptionStatus.ACTIVE)
47 quantity = Column(Integer, default=1)
48 current_period_start = Column(DateTime, nullable=False)
49 current_period_end = Column(DateTime, nullable=False)
50 cancel_at_period_end = Column(Boolean, default=False)
51 trial_ends_at = Column(DateTime, nullable=True)
52 cancelled_at = Column(DateTime, nullable=True)
53 created_at = Column(DateTime, server_default=func.now())
54 updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
55
56 plan = relationship("Plan")
57
58class UsageEvent(Base):
59 __tablename__ = "usage_events"
60
61 id = Column(String, primary_key=True)
62 idempotency_key = Column(String, unique=True, nullable=False)
63 subscription_id = Column(String, ForeignKey("subscriptions.id"), nullable=False)
64 metric_id = Column(String, nullable=False)
65 quantity = Column(BigInteger, nullable=False)
66 timestamp = Column(DateTime, nullable=False)
67 metadata = Column(JSON, default=dict)
68 created_at = Column(DateTime, server_default=func.now())
69
70class WebhookEvent(Base):
71 __tablename__ = "webhook_events"
72
73 stripe_event_id = Column(String, primary_key=True)
74 event_type = Column(String, nullable=False)
75 processed_at = Column(DateTime, server_default=func.now())
76
Billing Service
python
1import stripe
2from sqlalchemy.orm import Session
3from sqlalchemy import select, func as sa_func
4from datetime import datetime, timedelta
5from typing import Optional
6from uuid import uuid4
7from dataclasses import dataclass
8
9
10class ProrationResult:
11 credit_cents: int
12 charge_cents: int
13 net_cents: int
14 effective_at: datetime
15 description: str
16
17class BillingService:
18 def __init__(self, db: Session, stripe_api_key: str):
19 self.db = db
20 stripe.api_key = stripe_api_key
21
22 def create_subscription(
23 self,
24 customer_id: str,
25 plan_slug: str,
26 quantity: int,
27 stripe_customer_id: str,
28 ) -> Subscription:
29 plan = self.db.execute(
30 select(Plan).where(Plan.slug == plan_slug, Plan.active == True)
31 ).scalar_one()
32
33 stripe_sub = stripe.Subscription.create(
34 customer=stripe_customer_id,
35 items=[{"price": plan.stripe_price_id, "quantity": quantity}],
36 )
37
38 sub = Subscription(
39 id=str(uuid4()),
40 customer_id=customer_id,
41 plan_id=plan.id,
42 stripe_subscription_id=stripe_sub.id,
43 stripe_customer_id=stripe_customer_id,
44 status=SubscriptionStatus.ACTIVE,
45 quantity=quantity,
46 current_period_start=datetime.fromtimestamp(stripe_sub.current_period_start),
47 current_period_end=datetime.fromtimestamp(stripe_sub.current_period_end),
48 )
49
50 self.db.add(sub)
51 self.db.commit()
52 return sub
53
54 def change_plan(
55 self, subscription_id: str, new_plan_slug: str
56 ) -> tuple[Subscription, ProrationResult]:
57 sub = self.db.get(Subscription, subscription_id)
58 current_plan = self.db.get(Plan, sub.plan_id)
59 new_plan = self.db.execute(
60 select(Plan).where(Plan.slug == new_plan_slug)
61 ).scalar_one()
62
63 proration = self._calculate_proration(
64 current_plan, new_plan, sub.quantity, sub.quantity,
65 sub.current_period_start, sub.current_period_end,
66 )
67
68 stripe.Subscription.modify(
69 sub.stripe_subscription_id,
70 items=[{
71 "id": sub.stripe_subscription_id,
72 "price": new_plan.stripe_price_id,
73 }],
74 proration_behavior="create_prorations",
75 )
76
77 sub.plan_id = new_plan.id
78 self.db.commit()
79
80 return sub, proration
81
82 def update_seats(self, subscription_id: str, new_quantity: int) -> Subscription:
83 sub = self.db.get(Subscription, subscription_id)
84
85 stripe.Subscription.modify(
86 sub.stripe_subscription_id,
87 items=[{
88 "id": sub.stripe_subscription_id,
89 "quantity": new_quantity,
90 }],
91 proration_behavior="create_prorations",
92 )
93
94 sub.quantity = new_quantity
95 self.db.commit()
96 return sub
97
98 def cancel_subscription(
99 self, subscription_id: str, immediately: bool = False
100 ) -> Subscription:
101 sub = self.db.get(Subscription, subscription_id)
102
103 if immediately:
104 stripe.Subscription.cancel(sub.stripe_subscription_id)
105 sub.status = SubscriptionStatus.CANCELLED
106 sub.cancelled_at = datetime.utcnow()
107 else:
108 stripe.Subscription.modify(
109 sub.stripe_subscription_id,
110 cancel_at_period_end=True,
111 )
112 sub.cancel_at_period_end = True
113
114 self.db.commit()
115 return sub
116
117 def record_usage(
118 self,
119 subscription_id: str,
120 metric_id: str,
121 quantity: int,
122 idempotency_key: str,
123 timestamp: Optional[datetime] = None,
124 ) -> None:
125 existing = self.db.execute(
126 select(UsageEvent).where(UsageEvent.idempotency_key == idempotency_key)
127 ).scalar_one_or_none()
128
129 if existing:
130 return # Idempotent — skip duplicate
131
132 event = UsageEvent(
133 id=str(uuid4()),
134 idempotency_key=idempotency_key,
135 subscription_id=subscription_id,
136 metric_id=metric_id,
137 quantity=quantity,
138 timestamp=timestamp or datetime.utcnow(),
139 )
140 self.db.add(event)
141 self.db.commit()
142
143 def get_usage_summary(
144 self, subscription_id: str, metric_id: str,
145 period_start: datetime, period_end: datetime,
146 ) -> int:
147 result = self.db.execute(
148 select(sa_func.coalesce(sa_func.sum(UsageEvent.quantity), 0))
149 .where(
150 UsageEvent.subscription_id == subscription_id,
151 UsageEvent.metric_id == metric_id,
152 UsageEvent.timestamp >= period_start,
153 UsageEvent.timestamp < period_end,
154 )
155 ).scalar()
156 return int(result)
157
158 def check_feature_access(self, customer_id: str, feature: str) -> bool:
159 sub = self.db.execute(
160 select(Subscription)
161 .join(Plan)
162 .where(
163 Subscription.customer_id == customer_id,
164 Subscription.status == SubscriptionStatus.ACTIVE,
165 )
166 ).scalar_one_or_none()
167
168 if not sub:
169 return False
170 return feature in sub.plan.features
171
172 def _calculate_proration(
173 self, current_plan: Plan, new_plan: Plan,
174 current_qty: int, new_qty: int,
175 period_start: datetime, period_end: datetime,
176 ) -> ProrationResult:
177 now = datetime.utcnow()
178 total_days = (period_end - period_start).days
179 remaining_days = (period_end - now).days
180
181 current_total = current_plan.base_price_cents + (
182 current_plan.price_per_seat_cents * max(0, current_qty - current_plan.included_seats)
183 )
184 credit = round(current_total / total_days * remaining_days)
185
186 new_total = new_plan.base_price_cents + (
187 new_plan.price_per_seat_cents * max(0, new_qty - new_plan.included_seats)
188 )
189 charge = round(new_total / total_days * remaining_days)
190
191 return ProrationResult(
192 credit_cents=credit,
193 charge_cents=charge,
194 net_cents=charge - credit,
195 effective_at=now,
196 description=(
197 f"{current_plan.name} ({current_qty} seats) → "
198 f"{new_plan.name} ({new_qty} seats), "
199 f"{remaining_days} days remaining"
200 ),
201 )
202
Webhook Handler
python
1from fastapi import APIRouter, Request, HTTPException, Header
2import stripe
3
4router = APIRouter()
5
6
7async def handle_stripe_webhook(
8 request: Request,
9 stripe_signature: str = Header(alias="Stripe-Signature"),
10):
11 payload = await request.body()
12
13 try:
14 event = stripe.Webhook.construct_event(
15 payload, stripe_signature, STRIPE_WEBHOOK_SECRET
16 )
17 except stripe.error.SignatureVerificationError:
18 raise HTTPException(status_code=401, detail="Invalid signature")
19
20 # Check idempotency
21 existing = db.execute(
22 select(WebhookEvent).where(WebhookEvent.stripe_event_id == event.id)
23 ).scalar_one_or_none()
24
25 if existing:
26 return {"status": "already_processed"}
27
28 # Process event
29 handler_map = {
30 "customer.subscription.updated": handle_subscription_updated,
31 "customer.subscription.deleted": handle_subscription_deleted,
32 "invoice.payment_failed": handle_payment_failed,
33 "invoice.paid": handle_invoice_paid,
34 }
35
36 handler = handler_map.get(event.type)
37 if handler:
38 await handler(event.data.object)
39
40 # Mark processed
41 db.add(WebhookEvent(
42 stripe_event_id=event.id,
43 event_type=event.type,
44 ))
45 db.commit()
46
47 return {"status": "processed"}
48
49async def handle_subscription_updated(data: dict):
50 sub = db.execute(
51 select(Subscription).where(
52 Subscription.stripe_subscription_id == data["id"]
53 )
54 ).scalar_one_or_none()
55
56 if not sub:
57 return
58
59 status_map = {
60 "active": SubscriptionStatus.ACTIVE,
61 "trialing": SubscriptionStatus.TRIALING,
62 "past_due": SubscriptionStatus.PAST_DUE,
63 "canceled": SubscriptionStatus.CANCELLED,
64 }
65
66 sub.status = status_map.get(data["status"], SubscriptionStatus.ACTIVE)
67 sub.quantity = data["items"]["data"][0]["quantity"]
68 sub.current_period_start = datetime.fromtimestamp(data["current_period_start"])
69 sub.current_period_end = datetime.fromtimestamp(data["current_period_end"])
70 sub.cancel_at_period_end = data.get("cancel_at_period_end", False)
71 db.commit()
72
73async def handle_payment_failed(data: dict):
74 if not data.get("subscription"):
75 return
76
77 sub = db.execute(
78 select(Subscription).where(
79 Subscription.stripe_subscription_id == data["subscription"]
80 )
81 ).scalar_one_or_none()
82
83 if sub:
84 sub.status = SubscriptionStatus.PAST_DUE
85 db.commit()
86
Need a second opinion on your saas engineering architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallAPI Endpoints
python
1from fastapi import APIRouter, Depends, HTTPException
2from pydantic import BaseModel
3
4router = APIRouter(prefix="/api/billing")
5
6class CreateSubscriptionRequest(BaseModel):
7 plan_slug: str
8 quantity: int = 1
9
10class ChangePlanRequest(BaseModel):
11 plan_slug: str
12
13class UpdateSeatsRequest(BaseModel):
14 quantity: int
15
16
17def get_subscription(user_id: str = Depends(get_current_user)):
18 billing = BillingService(db, STRIPE_API_KEY)
19 sub = db.execute(
20 select(Subscription)
21 .join(Plan)
22 .where(
23 Subscription.customer_id == user_id,
24 Subscription.status.in_([
25 SubscriptionStatus.ACTIVE,
26 SubscriptionStatus.TRIALING,
27 ]),
28 )
29 ).scalar_one_or_none()
30
31 if not sub:
32 raise HTTPException(404, "No active subscription")
33
34 return {
35 "id": sub.id,
36 "plan": sub.plan.name,
37 "status": sub.status.value,
38 "seats": sub.quantity,
39 "current_period_end": sub.current_period_end.isoformat(),
40 "cancel_at_period_end": sub.cancel_at_period_end,
41 }
42
43
44def create_subscription(
45 req: CreateSubscriptionRequest,
46 user_id: str = Depends(get_current_user),
47):
48 billing = BillingService(db, STRIPE_API_KEY)
49 customer = get_or_create_stripe_customer(user_id)
50 sub = billing.create_subscription(
51 customer_id=user_id,
52 plan_slug=req.plan_slug,
53 quantity=req.quantity,
54 stripe_customer_id=customer.id,
55 )
56 return {"subscription_id": sub.id, "status": sub.status.value}
57
58
59def change_plan(
60 req: ChangePlanRequest,
61 user_id: str = Depends(get_current_user),
62):
63 billing = BillingService(db, STRIPE_API_KEY)
64 sub = get_active_subscription(user_id)
65 updated, proration = billing.change_plan(sub.id, req.plan_slug)
66 return {
67 "subscription": {"id": updated.id, "plan_id": updated.plan_id},
68 "proration": {
69 "credit_cents": proration.credit_cents,
70 "charge_cents": proration.charge_cents,
71 "net_cents": proration.net_cents,
72 },
73 }
74
75
76def update_seats(
77 req: UpdateSeatsRequest,
78 user_id: str = Depends(get_current_user),
79):
80 billing = BillingService(db, STRIPE_API_KEY)
81 sub = get_active_subscription(user_id)
82 updated = billing.update_seats(sub.id, req.quantity)
83 return {"seats": updated.quantity}
84
85
86def cancel_subscription(
87 user_id: str = Depends(get_current_user),
88 immediately: bool = False,
89):
90 billing = BillingService(db, STRIPE_API_KEY)
91 sub = get_active_subscription(user_id)
92 updated = billing.cancel_subscription(sub.id, immediately)
93 return {"status": updated.status.value}
94
Testing
python
1import pytest
2from unittest.mock import MagicMock, patch
3from datetime import datetime, timedelta
4
5class TestBillingService:
6 def setup_method(self):
7 self.db = create_test_session()
8 self.service = BillingService(self.db, "sk_test_fake")
9
10 # Seed a plan
11 plan = Plan(
12 id="plan-1", name="Pro", slug="pro",
13 billing_interval=BillingInterval.MONTH,
14 base_price_cents=4900, price_per_seat_cents=1000,
15 included_seats=1, features=["analytics", "api"],
16 stripe_price_id="price_test",
17 )
18 self.db.add(plan)
19 self.db.commit()
20
21 def test_proration_upgrade(self):
22 current = Plan(
23 base_price_cents=4900, price_per_seat_cents=0,
24 included_seats=1, name="Starter",
25 )
26 new = Plan(
27 base_price_cents=9900, price_per_seat_cents=0,
28 included_seats=5, name="Pro",
29 )
30
31 now = datetime.utcnow()
32 period_start = now - timedelta(days=15)
33 period_end = now + timedelta(days=15)
34
35 result = self.service._calculate_proration(
36 current, new, 1, 1, period_start, period_end
37 )
38
39 assert result.charge_cents > result.credit_cents
40 assert result.net_cents > 0
41
42 def test_usage_idempotency(self):
43 sub = create_test_subscription(self.db)
44
45 self.service.record_usage(sub.id, "api_calls", 100, "key-1")
46 self.service.record_usage(sub.id, "api_calls", 100, "key-1") # Duplicate
47
48 total = self.service.get_usage_summary(
49 sub.id, "api_calls",
50 sub.current_period_start, sub.current_period_end,
51 )
52 assert total == 100 # Not 200
53
54 def test_feature_access(self):
55 sub = create_test_subscription(self.db, plan_slug="pro")
56
57 assert self.service.check_feature_access(sub.customer_id, "analytics") is True
58 assert self.service.check_feature_access(sub.customer_id, "enterprise_sso") is False
59
Conclusion
Python's billing implementation benefits from SQLAlchemy's query builder for complex billing queries, Pydantic for request validation, and the Stripe Python SDK's clean API mapping. The service layer pattern keeps billing logic testable — mock the Stripe SDK for unit tests and use a test database for integration tests.
The critical implementation details are monetary amounts in cents (avoiding floating-point), idempotent usage recording, and atomic subscription state updates. Every webhook handler must check for prior processing before mutating state, and every Stripe API call should use idempotency keys to prevent duplicate operations during retries.
For production, add Celery tasks for asynchronous operations (dunning emails, usage aggregation, invoice generation) and Redis for caching plan definitions and feature flags. The webhook endpoint should return 200 immediately and enqueue processing to avoid Stripe's 20-second timeout.