Coverage for src/ai_lls_lib/payment/stripe_manager.py: 58%
212 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 17:06 +0000
1"""Stripe API management with metadata conventions."""
3import logging
4import os
5from typing import Any, cast
7try:
8 import stripe
10 HAS_STRIPE = True
11except ImportError:
12 stripe = None # type: ignore[assignment]
13 HAS_STRIPE = False
15from .models import Plan
17logger = logging.getLogger(__name__)
20class StripeManager:
21 """
22 Manages Stripe resources with metadata conventions.
23 Uses metadata to discover and filter products/prices.
24 """
26 METADATA_SCHEMA = {
27 "product_type": "landline_scrubber",
28 "environment": None, # Set at runtime
29 "tier": None,
30 "credits": None,
31 "active": "true",
32 }
34 def __init__(self, api_key: str | None = None, environment: str | None = None):
35 """Initialize with Stripe API key and environment."""
36 if not HAS_STRIPE or not stripe:
37 raise ImportError("stripe package not installed. Run: pip install stripe")
39 self.api_key = api_key or os.environ.get("STRIPE_SECRET_KEY")
40 if not self.api_key:
41 raise ValueError("Stripe API key not provided and STRIPE_SECRET_KEY not set")
43 stripe.api_key = self.api_key
44 self.environment = environment or os.environ.get("ENVIRONMENT", "staging")
46 def list_plans(self) -> list[Plan]:
47 """
48 Fetch active plans from Stripe using metadata.
49 Returns list of Plan objects sorted by price.
50 """
51 try:
52 # Fetch all active prices with expanded product data
53 prices = stripe.Price.list(active=True, expand=["data.product"], limit=100)
55 plans = []
56 for price in prices.data:
57 metadata = cast(dict[str, str], price.metadata or {})
59 # Filter by our metadata conventions
60 if (
61 metadata["product_type"] == "landline_scrubber"
62 if "product_type" in metadata
63 else False
64 ) and (metadata["active"] == "true" if "active" in metadata else False):
65 # Convert to Plan object
66 # price.product is expanded to Product object due to expand param
67 product = price.product
68 if isinstance(product, str):
69 # Shouldn't happen with expand, but handle gracefully
70 continue
71 plan = Plan.from_stripe_price(price, product) # type: ignore[arg-type]
72 plans.append(plan)
74 # Sort by price amount
75 plans.sort(key=lambda p: p.plan_amount)
77 logger.info(f"Found {len(plans)} active plans for environment {self.environment}")
78 return plans
80 except stripe.error.StripeError as e:
81 logger.error(f"Stripe error listing plans: {e}")
82 # Let error propagate - no fallback to mock data
83 raise
85 def create_setup_intent(self, user_id: str) -> dict[str, str]:
86 """
87 Create a SetupIntent for secure payment method collection.
88 Frontend will confirm this with Stripe Elements.
89 """
90 try:
91 # Get or create customer
92 customer = self._get_or_create_customer(user_id)
94 # Create SetupIntent
95 setup_intent = stripe.SetupIntent.create(
96 customer=customer.id, metadata={"user_id": user_id, "environment": self.environment}
97 )
99 return {
100 "client_secret": setup_intent.client_secret,
101 "setup_intent_id": setup_intent.id,
102 "customer_id": customer.id,
103 }
105 except stripe.error.StripeError as e:
106 logger.error(f"Stripe error creating setup intent: {e}")
107 raise
109 def attach_payment_method(
110 self, user_id: str, payment_method_id: str, billing_details: dict[str, Any]
111 ) -> dict[str, Any]:
112 """
113 Attach a payment method to customer (legacy path).
114 Returns whether this is the first card.
115 """
116 try:
117 # Get or create customer
118 customer = self._get_or_create_customer(user_id)
120 # Attach payment method to customer
121 stripe.PaymentMethod.attach(payment_method_id, customer=customer.id)
123 # Update billing details if provided
124 if billing_details:
125 stripe.PaymentMethod.modify(
126 payment_method_id,
127 billing_details=billing_details, # type: ignore[arg-type]
128 )
130 # Check if this is the first payment method
131 payment_methods = stripe.PaymentMethod.list(customer=customer.id, type="card")
133 first_card = len(payment_methods.data) == 1
135 # Set as default if first card
136 if first_card:
137 stripe.Customer.modify(
138 customer.id, invoice_settings={"default_payment_method": payment_method_id}
139 )
141 return {
142 "payment_method_id": payment_method_id,
143 "first_card": first_card,
144 "customer_id": customer.id,
145 }
147 except stripe.error.StripeError as e:
148 logger.error(f"Stripe error attaching payment method: {e}")
149 raise
151 def verify_payment_method(self, user_id: str, payment_method_id: str) -> dict[str, Any]:
152 """
153 Perform $1 verification charge on new payment method.
154 """
155 try:
156 customer = self._get_or_create_customer(user_id)
158 # Create $1 verification charge
159 payment_intent = stripe.PaymentIntent.create(
160 amount=100, # $1.00 in cents
161 currency="usd",
162 customer=customer.id,
163 payment_method=payment_method_id,
164 off_session=True,
165 confirm=True,
166 description="Card verification - $1 charge",
167 metadata={
168 "user_id": user_id,
169 "type": "verification",
170 "environment": self.environment,
171 },
172 )
174 return {"status": payment_intent.status, "payment_intent_id": payment_intent.id}
176 except stripe.error.StripeError as e:
177 logger.error(f"Stripe error verifying payment method: {e}")
178 raise
180 def create_subscription_checkout(
181 self,
182 price_id: str,
183 user_id: str,
184 success_url: str,
185 cancel_url: str,
186 customer_id: str | None = None,
187 ) -> Any:
188 """Create a Stripe Checkout Session in subscription mode.
190 The mirror image of charge_prepaid's guard: only recurring prices are accepted
191 here. user_id is set in subscription metadata so the subscription webhooks can
192 write subscription_status for the right account -- without it a paid
193 subscription is never fulfilled.
195 Returns the Checkout Session (id and url are what callers redirect with).
196 """
197 price = stripe.Price.retrieve(price_id)
198 if not getattr(price, "recurring", None):
199 raise ValueError(
200 f"Price {price_id} is not a subscription price; use the one-time purchase path"
201 )
203 if customer_id is None:
204 customer_id = self._get_or_create_customer(user_id).id
206 session = stripe.checkout.Session.create(
207 mode="subscription",
208 customer=customer_id,
209 line_items=[{"price": price_id, "quantity": 1}],
210 success_url=success_url,
211 cancel_url=cancel_url,
212 subscription_data={"metadata": {"user_id": user_id, "environment": self.environment}},
213 metadata={"user_id": user_id, "environment": self.environment},
214 )
215 logger.info(f"Created subscription checkout session {session.id} for user {user_id}")
216 return session
218 def charge_prepaid(
219 self,
220 user_id: str,
221 reference_code: str,
222 amount: float | None = None,
223 stripe_customer_id: str | None = None,
224 ) -> dict[str, Any]:
225 """
226 Charge saved payment method for credit purchase.
227 Supports both fixed-price and metadata-based variable-amount plans.
229 Pass stripe_customer_id when the caller has it stored. Without it the customer
230 is found with Customer.search, which is eventually consistent: a purchase made
231 seconds after the first card was added can miss the customer, create a second
232 one with no card, and fail with "No payment method on file".
233 """
234 try:
235 customer = self._get_or_create_customer(user_id, stripe_customer_id=stripe_customer_id)
237 # Look up price from Stripe
238 prices = stripe.Price.list(active=True, limit=100, expand=["data.product"])
239 price = None
241 for p in prices.data:
242 metadata = cast(dict[str, str], p.metadata or {})
243 # Match by price ID or plan_reference in metadata
244 plan_reference = (
245 metadata["plan_reference"] if "plan_reference" in metadata else None
246 )
247 tier = metadata["tier"] if "tier" in metadata else None
248 env = metadata["environment"] if "environment" in metadata else None
249 if (
250 p.id == reference_code
251 or plan_reference == reference_code
252 or (tier == reference_code and env == self.environment)
253 ):
254 price = p
255 break
257 if not price:
258 raise ValueError(f"Invalid plan reference: {reference_code}")
260 # Fail closed on subscription prices. This path creates a one-time PaymentIntent;
261 # charging a recurring price here takes the money and creates no subscription.
262 # The check runs before the variable-amount branch so no price shape can bypass it.
263 if getattr(price, "recurring", None):
264 raise ValueError(
265 f"Plan {reference_code} is a subscription and cannot be charged as a "
266 "one-time purchase; use the subscription checkout"
267 )
269 price_metadata = cast(dict[str, str], price.metadata or {})
271 # Check if this is a variable amount plan
272 variable_amount = (
273 price_metadata["variable_amount"] if "variable_amount" in price_metadata else None
274 )
275 if variable_amount == "true":
276 # Variable amount plan - validate amount
277 if not amount:
278 raise ValueError("Amount required for variable-amount plan")
280 # Get validation rules from metadata
281 min_amount = float(
282 price_metadata["min_amount"] if "min_amount" in price_metadata else "5"
283 )
284 if amount < min_amount:
285 raise ValueError(f"Amount ${amount} is below minimum ${min_amount}")
287 # Check against default amounts if specified
288 default_amounts_str = (
289 price_metadata["default_amounts"] if "default_amounts" in price_metadata else ""
290 )
291 if default_amounts_str:
292 allowed_amounts = [float(x.strip()) for x in default_amounts_str.split(",")]
293 # Allow default amounts OR any amount >= minimum
294 if amount not in allowed_amounts and amount < max(allowed_amounts):
295 logger.info(
296 f"Amount ${amount} not in defaults {allowed_amounts}, but allowed as >= ${min_amount}"
297 )
299 # Calculate credits based on credits_per_dollar
300 credits_per_dollar = float(
301 price_metadata["credits_per_dollar"]
302 if "credits_per_dollar" in price_metadata
303 else "285"
304 )
305 credits_to_add = int(amount * credits_per_dollar)
306 charge_amount = int(amount * 100) # Convert to cents
308 else:
309 # Fixed price plan
310 charge_amount = price.unit_amount or 0
311 credits_str = price_metadata["credits"] if "credits" in price_metadata else "0"
312 if credits_str.lower() == "unlimited":
313 credits_to_add = 0 # Subscription handles this differently
314 else:
315 credits_to_add = int(credits_str)
317 # Get default payment method
318 invoice_settings = customer.invoice_settings
319 default_pm = (
320 invoice_settings["default_payment_method"]
321 if invoice_settings and "default_payment_method" in invoice_settings
322 else None
323 )
324 if not default_pm:
325 # Try to get first payment method
326 payment_methods = stripe.PaymentMethod.list(
327 customer=customer.id, type="card", limit=1
328 )
329 if not payment_methods.data:
330 raise ValueError("No payment method on file")
331 default_pm = payment_methods.data[0].id
333 # Create payment intent
334 payment_intent = stripe.PaymentIntent.create(
335 amount=charge_amount,
336 currency="usd",
337 customer=customer.id,
338 payment_method=default_pm,
339 off_session=True,
340 confirm=True,
341 description=f"Credit purchase - {credits_to_add} credits",
342 metadata={
343 "user_id": user_id,
344 "credits": str(credits_to_add),
345 "reference_code": reference_code,
346 "environment": self.environment,
347 },
348 )
350 return {
351 "id": payment_intent.id,
352 "status": payment_intent.status,
353 "credits_added": credits_to_add,
354 "amount_charged": charge_amount / 100, # Convert back to dollars
355 }
357 except stripe.error.StripeError as e:
358 logger.error(f"Stripe error processing payment: {e}")
359 raise
361 def customer_has_payment_method(self, stripe_customer_id: str) -> bool:
362 """
363 Check if customer has any saved payment methods.
364 """
365 try:
366 payment_methods = stripe.PaymentMethod.list(
367 customer=stripe_customer_id, type="card", limit=1
368 )
369 return len(payment_methods.data) > 0
370 except stripe.error.StripeError as e:
371 logger.error(f"Stripe error checking payment methods: {e}")
372 return False
374 def list_payment_methods(self, stripe_customer_id: str) -> dict[str, Any]:
375 """
376 List all payment methods for a customer.
377 """
378 try:
379 # Get customer to find default payment method
380 customer = stripe.Customer.retrieve(stripe_customer_id)
381 invoice_settings = customer.invoice_settings
382 default_pm_id = invoice_settings.default_payment_method if invoice_settings else None
384 # List all payment methods
385 payment_methods = stripe.PaymentMethod.list(customer=stripe_customer_id, type="card")
387 items = []
388 for pm in payment_methods.data:
389 card = pm.card
390 if card:
391 items.append(
392 {
393 "id": pm.id,
394 "brand": card.brand,
395 "last4": card.last4,
396 "exp_month": card.exp_month,
397 "exp_year": card.exp_year,
398 "is_default": pm.id == default_pm_id,
399 }
400 )
402 return {"items": items, "default_payment_method_id": default_pm_id}
404 except stripe.error.StripeError as e:
405 logger.error(f"Stripe error listing payment methods: {e}")
406 return {"items": [], "default_payment_method_id": None}
408 def _get_or_create_customer(
409 self, user_id: str, email: str | None = None, stripe_customer_id: str | None = None
410 ) -> Any:
411 """
412 Get existing Stripe customer or create new one.
413 A known customer id is retrieved directly (strongly consistent). Otherwise
414 checks by user_id in metadata, then by email if provided.
415 """
416 try:
417 if stripe_customer_id:
418 known = stripe.Customer.retrieve(stripe_customer_id)
419 if not getattr(known, "deleted", False):
420 return known
422 # First try to find by user_id in metadata
423 search_results = stripe.Customer.search(
424 query=f'metadata["user_id"]:"{user_id}"', limit=1
425 )
427 if search_results.data:
428 return search_results.data[0]
430 # If email provided, try to find by email
431 if email:
432 email_results = stripe.Customer.list(email=email, limit=1)
433 if email_results.data:
434 # Update metadata with user_id
435 customer = email_results.data[0]
436 stripe.Customer.modify(customer.id, metadata={"user_id": user_id})
437 return customer
439 # Create new customer (email is optional)
440 create_params: dict[str, Any] = {
441 "metadata": {"user_id": user_id, "environment": self.environment}
442 }
443 if email:
444 create_params["email"] = email
445 return stripe.Customer.create(**create_params)
447 except stripe.error.StripeError as e:
448 logger.error(f"Stripe error getting/creating customer: {e}")
449 raise
451 def create_subscription(self, user_id: str, email: str, price_id: str) -> dict[str, Any]:
452 """Create a subscription for unlimited access."""
453 try:
454 # Create or retrieve customer
455 customers = stripe.Customer.list(email=email, limit=1)
456 if customers.data:
457 customer = customers.data[0]
458 else:
459 customer = stripe.Customer.create(email=email, metadata={"user_id": user_id})
461 # Create subscription
462 subscription = stripe.Subscription.create(
463 customer=customer.id,
464 items=[{"price": price_id}],
465 metadata={"user_id": user_id, "environment": self.environment},
466 )
468 return {
469 "subscription_id": subscription.id,
470 "status": subscription.status,
471 "customer_id": customer.id,
472 }
474 except stripe.error.StripeError as e:
475 logger.error(f"Stripe error creating subscription: {e}")
476 raise
478 def pause_subscription(self, subscription_id: str) -> dict[str, str]:
479 """Pause a subscription."""
480 try:
481 stripe.Subscription.modify(
482 subscription_id, pause_collection={"behavior": "mark_uncollectible"}
483 )
484 return {"message": "Subscription paused", "status": "paused"}
485 except stripe.error.StripeError as e:
486 logger.error(f"Stripe error pausing subscription: {e}")
487 raise
489 def resume_subscription(self, subscription_id: str) -> dict[str, str]:
490 """Resume a paused subscription."""
491 try:
492 stripe.Subscription.modify(
493 subscription_id,
494 pause_collection="", # Remove pause
495 )
496 return {"message": "Subscription resumed", "status": "active"}
497 except stripe.error.StripeError as e:
498 logger.error(f"Stripe error resuming subscription: {e}")
499 raise
501 def cancel_subscription(self, subscription_id: str) -> dict[str, Any]:
502 """Cancel a subscription at the end of the period the customer has paid for.
504 The published refund policy gives no refunds or prorations for a cancelled
505 month, so the customer keeps access until the period ends. Stripe then ends the
506 subscription and sends customer.subscription.deleted, which records the
507 cancellation. An immediate cancellation with a refund is a staff action in the
508 Stripe dashboard and ends in the same event.
509 """
510 try:
511 subscription = stripe.Subscription.modify(subscription_id, cancel_at_period_end=True)
512 # Since Stripe API 2025-03-31 the billing period is on the subscription item.
513 item = subscription["items"]["data"][0]
514 return {
515 "message": "Subscription cancels at the end of the current period",
516 "status": subscription["status"],
517 "cancel_at_period_end": subscription["cancel_at_period_end"],
518 "current_period_end": item["current_period_end"],
519 }
520 except stripe.error.StripeError as e:
521 logger.error(f"Stripe error cancelling subscription: {e}")
522 raise
524 def create_billing_portal_session(self, customer_id: str, return_url: str) -> str:
525 """Create a Stripe Billing Portal session for the customer.
527 Returns the portal session URL.
528 """
529 try:
530 session = stripe.billing_portal.Session.create(
531 customer=customer_id,
532 return_url=return_url,
533 )
534 return session.url
535 except stripe.error.StripeError as e:
536 logger.error(f"Stripe error creating billing portal session: {e}")
537 raise