Coverage for src/ai_lls_lib/payment/stripe_manager.py: 55%
207 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-24 12:44 +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, user_id: str, reference_code: str, amount: float | None = None
220 ) -> dict[str, Any]:
221 """
222 Charge saved payment method for credit purchase.
223 Supports both fixed-price and metadata-based variable-amount plans.
224 """
225 try:
226 customer = self._get_or_create_customer(user_id)
228 # Look up price from Stripe
229 prices = stripe.Price.list(active=True, limit=100, expand=["data.product"])
230 price = None
232 for p in prices.data:
233 metadata = cast(dict[str, str], p.metadata or {})
234 # Match by price ID or plan_reference in metadata
235 plan_reference = (
236 metadata["plan_reference"] if "plan_reference" in metadata else None
237 )
238 tier = metadata["tier"] if "tier" in metadata else None
239 env = metadata["environment"] if "environment" in metadata else None
240 if (
241 p.id == reference_code
242 or plan_reference == reference_code
243 or (tier == reference_code and env == self.environment)
244 ):
245 price = p
246 break
248 if not price:
249 raise ValueError(f"Invalid plan reference: {reference_code}")
251 # Fail closed on subscription prices. This path creates a one-time PaymentIntent;
252 # charging a recurring price here takes the money and creates no subscription.
253 # The check runs before the variable-amount branch so no price shape can bypass it.
254 if getattr(price, "recurring", None):
255 raise ValueError(
256 f"Plan {reference_code} is a subscription and cannot be charged as a "
257 "one-time purchase; use the subscription checkout"
258 )
260 price_metadata = cast(dict[str, str], price.metadata or {})
262 # Check if this is a variable amount plan
263 variable_amount = (
264 price_metadata["variable_amount"] if "variable_amount" in price_metadata else None
265 )
266 if variable_amount == "true":
267 # Variable amount plan - validate amount
268 if not amount:
269 raise ValueError("Amount required for variable-amount plan")
271 # Get validation rules from metadata
272 min_amount = float(
273 price_metadata["min_amount"] if "min_amount" in price_metadata else "5"
274 )
275 if amount < min_amount:
276 raise ValueError(f"Amount ${amount} is below minimum ${min_amount}")
278 # Check against default amounts if specified
279 default_amounts_str = (
280 price_metadata["default_amounts"] if "default_amounts" in price_metadata else ""
281 )
282 if default_amounts_str:
283 allowed_amounts = [float(x.strip()) for x in default_amounts_str.split(",")]
284 # Allow default amounts OR any amount >= minimum
285 if amount not in allowed_amounts and amount < max(allowed_amounts):
286 logger.info(
287 f"Amount ${amount} not in defaults {allowed_amounts}, but allowed as >= ${min_amount}"
288 )
290 # Calculate credits based on credits_per_dollar
291 credits_per_dollar = float(
292 price_metadata["credits_per_dollar"]
293 if "credits_per_dollar" in price_metadata
294 else "285"
295 )
296 credits_to_add = int(amount * credits_per_dollar)
297 charge_amount = int(amount * 100) # Convert to cents
299 else:
300 # Fixed price plan
301 charge_amount = price.unit_amount or 0
302 credits_str = price_metadata["credits"] if "credits" in price_metadata else "0"
303 if credits_str.lower() == "unlimited":
304 credits_to_add = 0 # Subscription handles this differently
305 else:
306 credits_to_add = int(credits_str)
308 # Get default payment method
309 invoice_settings = customer.invoice_settings
310 default_pm = (
311 invoice_settings["default_payment_method"]
312 if invoice_settings and "default_payment_method" in invoice_settings
313 else None
314 )
315 if not default_pm:
316 # Try to get first payment method
317 payment_methods = stripe.PaymentMethod.list(
318 customer=customer.id, type="card", limit=1
319 )
320 if not payment_methods.data:
321 raise ValueError("No payment method on file")
322 default_pm = payment_methods.data[0].id
324 # Create payment intent
325 payment_intent = stripe.PaymentIntent.create(
326 amount=charge_amount,
327 currency="usd",
328 customer=customer.id,
329 payment_method=default_pm,
330 off_session=True,
331 confirm=True,
332 description=f"Credit purchase - {credits_to_add} credits",
333 metadata={
334 "user_id": user_id,
335 "credits": str(credits_to_add),
336 "reference_code": reference_code,
337 "environment": self.environment,
338 },
339 )
341 return {
342 "id": payment_intent.id,
343 "status": payment_intent.status,
344 "credits_added": credits_to_add,
345 "amount_charged": charge_amount / 100, # Convert back to dollars
346 }
348 except stripe.error.StripeError as e:
349 logger.error(f"Stripe error processing payment: {e}")
350 raise
352 def customer_has_payment_method(self, stripe_customer_id: str) -> bool:
353 """
354 Check if customer has any saved payment methods.
355 """
356 try:
357 payment_methods = stripe.PaymentMethod.list(
358 customer=stripe_customer_id, type="card", limit=1
359 )
360 return len(payment_methods.data) > 0
361 except stripe.error.StripeError as e:
362 logger.error(f"Stripe error checking payment methods: {e}")
363 return False
365 def list_payment_methods(self, stripe_customer_id: str) -> dict[str, Any]:
366 """
367 List all payment methods for a customer.
368 """
369 try:
370 # Get customer to find default payment method
371 customer = stripe.Customer.retrieve(stripe_customer_id)
372 invoice_settings = customer.invoice_settings
373 default_pm_id = invoice_settings.default_payment_method if invoice_settings else None
375 # List all payment methods
376 payment_methods = stripe.PaymentMethod.list(customer=stripe_customer_id, type="card")
378 items = []
379 for pm in payment_methods.data:
380 card = pm.card
381 if card:
382 items.append(
383 {
384 "id": pm.id,
385 "brand": card.brand,
386 "last4": card.last4,
387 "exp_month": card.exp_month,
388 "exp_year": card.exp_year,
389 "is_default": pm.id == default_pm_id,
390 }
391 )
393 return {"items": items, "default_payment_method_id": default_pm_id}
395 except stripe.error.StripeError as e:
396 logger.error(f"Stripe error listing payment methods: {e}")
397 return {"items": [], "default_payment_method_id": None}
399 def _get_or_create_customer(self, user_id: str, email: str | None = None) -> Any:
400 """
401 Get existing Stripe customer or create new one.
402 First checks by user_id in metadata, then by email if provided.
403 """
404 try:
405 # First try to find by user_id in metadata
406 search_results = stripe.Customer.search(
407 query=f'metadata["user_id"]:"{user_id}"', limit=1
408 )
410 if search_results.data:
411 return search_results.data[0]
413 # If email provided, try to find by email
414 if email:
415 email_results = stripe.Customer.list(email=email, limit=1)
416 if email_results.data:
417 # Update metadata with user_id
418 customer = email_results.data[0]
419 stripe.Customer.modify(customer.id, metadata={"user_id": user_id})
420 return customer
422 # Create new customer (email is optional)
423 create_params: dict[str, Any] = {
424 "metadata": {"user_id": user_id, "environment": self.environment}
425 }
426 if email:
427 create_params["email"] = email
428 return stripe.Customer.create(**create_params)
430 except stripe.error.StripeError as e:
431 logger.error(f"Stripe error getting/creating customer: {e}")
432 raise
434 def create_subscription(self, user_id: str, email: str, price_id: str) -> dict[str, Any]:
435 """Create a subscription for unlimited access."""
436 try:
437 # Create or retrieve customer
438 customers = stripe.Customer.list(email=email, limit=1)
439 if customers.data:
440 customer = customers.data[0]
441 else:
442 customer = stripe.Customer.create(email=email, metadata={"user_id": user_id})
444 # Create subscription
445 subscription = stripe.Subscription.create(
446 customer=customer.id,
447 items=[{"price": price_id}],
448 metadata={"user_id": user_id, "environment": self.environment},
449 )
451 return {
452 "subscription_id": subscription.id,
453 "status": subscription.status,
454 "customer_id": customer.id,
455 }
457 except stripe.error.StripeError as e:
458 logger.error(f"Stripe error creating subscription: {e}")
459 raise
461 def pause_subscription(self, subscription_id: str) -> dict[str, str]:
462 """Pause a subscription."""
463 try:
464 stripe.Subscription.modify(
465 subscription_id, pause_collection={"behavior": "mark_uncollectible"}
466 )
467 return {"message": "Subscription paused", "status": "paused"}
468 except stripe.error.StripeError as e:
469 logger.error(f"Stripe error pausing subscription: {e}")
470 raise
472 def resume_subscription(self, subscription_id: str) -> dict[str, str]:
473 """Resume a paused subscription."""
474 try:
475 stripe.Subscription.modify(
476 subscription_id,
477 pause_collection="", # Remove pause
478 )
479 return {"message": "Subscription resumed", "status": "active"}
480 except stripe.error.StripeError as e:
481 logger.error(f"Stripe error resuming subscription: {e}")
482 raise
484 def cancel_subscription(self, subscription_id: str) -> dict[str, str]:
485 """Cancel a subscription."""
486 try:
487 stripe.Subscription.cancel(subscription_id)
488 return {"message": "Subscription cancelled", "status": "cancelled"}
489 except stripe.error.StripeError as e:
490 logger.error(f"Stripe error cancelling subscription: {e}")
491 raise
493 def create_billing_portal_session(self, customer_id: str, return_url: str) -> str:
494 """Create a Stripe Billing Portal session for the customer.
496 Returns the portal session URL.
497 """
498 try:
499 session = stripe.billing_portal.Session.create(
500 customer=customer_id,
501 return_url=return_url,
502 )
503 return session.url
504 except stripe.error.StripeError as e:
505 logger.error(f"Stripe error creating billing portal session: {e}")
506 raise