Coverage for src / ai_lls_lib / payment / stripe_manager.py: 52%
196 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-06 23:45 +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 charge_prepaid(
181 self, user_id: str, reference_code: str, amount: float | None = None
182 ) -> dict[str, Any]:
183 """
184 Charge saved payment method for credit purchase.
185 Supports both fixed-price and metadata-based variable-amount plans.
186 """
187 try:
188 customer = self._get_or_create_customer(user_id)
190 # Look up price from Stripe
191 prices = stripe.Price.list(active=True, limit=100, expand=["data.product"])
192 price = None
194 for p in prices.data:
195 metadata = cast(dict[str, str], p.metadata or {})
196 # Match by price ID or plan_reference in metadata
197 plan_reference = (
198 metadata["plan_reference"] if "plan_reference" in metadata else None
199 )
200 tier = metadata["tier"] if "tier" in metadata else None
201 env = metadata["environment"] if "environment" in metadata else None
202 if (
203 p.id == reference_code
204 or plan_reference == reference_code
205 or (tier == reference_code and env == self.environment)
206 ):
207 price = p
208 break
210 if not price:
211 raise ValueError(f"Invalid plan reference: {reference_code}")
213 price_metadata = cast(dict[str, str], price.metadata or {})
215 # Check if this is a variable amount plan
216 variable_amount = (
217 price_metadata["variable_amount"] if "variable_amount" in price_metadata else None
218 )
219 if variable_amount == "true":
220 # Variable amount plan - validate amount
221 if not amount:
222 raise ValueError("Amount required for variable-amount plan")
224 # Get validation rules from metadata
225 min_amount = float(
226 price_metadata["min_amount"] if "min_amount" in price_metadata else "5"
227 )
228 if amount < min_amount:
229 raise ValueError(f"Amount ${amount} is below minimum ${min_amount}")
231 # Check against default amounts if specified
232 default_amounts_str = (
233 price_metadata["default_amounts"] if "default_amounts" in price_metadata else ""
234 )
235 if default_amounts_str:
236 allowed_amounts = [float(x.strip()) for x in default_amounts_str.split(",")]
237 # Allow default amounts OR any amount >= minimum
238 if amount not in allowed_amounts and amount < max(allowed_amounts):
239 logger.info(
240 f"Amount ${amount} not in defaults {allowed_amounts}, but allowed as >= ${min_amount}"
241 )
243 # Calculate credits based on credits_per_dollar
244 credits_per_dollar = float(
245 price_metadata["credits_per_dollar"]
246 if "credits_per_dollar" in price_metadata
247 else "285"
248 )
249 credits_to_add = int(amount * credits_per_dollar)
250 charge_amount = int(amount * 100) # Convert to cents
252 else:
253 # Fixed price plan
254 charge_amount = price.unit_amount or 0
255 credits_str = price_metadata["credits"] if "credits" in price_metadata else "0"
256 if credits_str.lower() == "unlimited":
257 credits_to_add = 0 # Subscription handles this differently
258 else:
259 credits_to_add = int(credits_str)
261 # Get default payment method
262 invoice_settings = customer.invoice_settings
263 default_pm = (
264 invoice_settings["default_payment_method"]
265 if invoice_settings and "default_payment_method" in invoice_settings
266 else None
267 )
268 if not default_pm:
269 # Try to get first payment method
270 payment_methods = stripe.PaymentMethod.list(
271 customer=customer.id, type="card", limit=1
272 )
273 if not payment_methods.data:
274 raise ValueError("No payment method on file")
275 default_pm = payment_methods.data[0].id
277 # Create payment intent
278 payment_intent = stripe.PaymentIntent.create(
279 amount=charge_amount,
280 currency="usd",
281 customer=customer.id,
282 payment_method=default_pm,
283 off_session=True,
284 confirm=True,
285 description=f"Credit purchase - {credits_to_add} credits",
286 metadata={
287 "user_id": user_id,
288 "credits": str(credits_to_add),
289 "reference_code": reference_code,
290 "environment": self.environment,
291 },
292 )
294 return {
295 "id": payment_intent.id,
296 "status": payment_intent.status,
297 "credits_added": credits_to_add,
298 "amount_charged": charge_amount / 100, # Convert back to dollars
299 }
301 except stripe.error.StripeError as e:
302 logger.error(f"Stripe error processing payment: {e}")
303 raise
305 def customer_has_payment_method(self, stripe_customer_id: str) -> bool:
306 """
307 Check if customer has any saved payment methods.
308 """
309 try:
310 payment_methods = stripe.PaymentMethod.list(
311 customer=stripe_customer_id, type="card", limit=1
312 )
313 return len(payment_methods.data) > 0
314 except stripe.error.StripeError as e:
315 logger.error(f"Stripe error checking payment methods: {e}")
316 return False
318 def list_payment_methods(self, stripe_customer_id: str) -> dict[str, Any]:
319 """
320 List all payment methods for a customer.
321 """
322 try:
323 # Get customer to find default payment method
324 customer = stripe.Customer.retrieve(stripe_customer_id)
325 invoice_settings = customer.invoice_settings
326 default_pm_id = invoice_settings.default_payment_method if invoice_settings else None
328 # List all payment methods
329 payment_methods = stripe.PaymentMethod.list(customer=stripe_customer_id, type="card")
331 items = []
332 for pm in payment_methods.data:
333 card = pm.card
334 if card:
335 items.append(
336 {
337 "id": pm.id,
338 "brand": card.brand,
339 "last4": card.last4,
340 "exp_month": card.exp_month,
341 "exp_year": card.exp_year,
342 "is_default": pm.id == default_pm_id,
343 }
344 )
346 return {"items": items, "default_payment_method_id": default_pm_id}
348 except stripe.error.StripeError as e:
349 logger.error(f"Stripe error listing payment methods: {e}")
350 return {"items": [], "default_payment_method_id": None}
352 def _get_or_create_customer(self, user_id: str, email: str | None = None) -> Any:
353 """
354 Get existing Stripe customer or create new one.
355 First checks by user_id in metadata, then by email if provided.
356 """
357 try:
358 # First try to find by user_id in metadata
359 search_results = stripe.Customer.search(
360 query=f'metadata["user_id"]:"{user_id}"', limit=1
361 )
363 if search_results.data:
364 return search_results.data[0]
366 # If email provided, try to find by email
367 if email:
368 email_results = stripe.Customer.list(email=email, limit=1)
369 if email_results.data:
370 # Update metadata with user_id
371 customer = email_results.data[0]
372 stripe.Customer.modify(customer.id, metadata={"user_id": user_id})
373 return customer
375 # Create new customer (email is optional)
376 create_params: dict[str, Any] = {
377 "metadata": {"user_id": user_id, "environment": self.environment}
378 }
379 if email:
380 create_params["email"] = email
381 return stripe.Customer.create(**create_params)
383 except stripe.error.StripeError as e:
384 logger.error(f"Stripe error getting/creating customer: {e}")
385 raise
387 def create_subscription(self, user_id: str, email: str, price_id: str) -> dict[str, Any]:
388 """Create a subscription for unlimited access."""
389 try:
390 # Create or retrieve customer
391 customers = stripe.Customer.list(email=email, limit=1)
392 if customers.data:
393 customer = customers.data[0]
394 else:
395 customer = stripe.Customer.create(email=email, metadata={"user_id": user_id})
397 # Create subscription
398 subscription = stripe.Subscription.create(
399 customer=customer.id,
400 items=[{"price": price_id}],
401 metadata={"user_id": user_id, "environment": self.environment},
402 )
404 return {
405 "subscription_id": subscription.id,
406 "status": subscription.status,
407 "customer_id": customer.id,
408 }
410 except stripe.error.StripeError as e:
411 logger.error(f"Stripe error creating subscription: {e}")
412 raise
414 def pause_subscription(self, subscription_id: str) -> dict[str, str]:
415 """Pause a subscription."""
416 try:
417 stripe.Subscription.modify(
418 subscription_id, pause_collection={"behavior": "mark_uncollectible"}
419 )
420 return {"message": "Subscription paused", "status": "paused"}
421 except stripe.error.StripeError as e:
422 logger.error(f"Stripe error pausing subscription: {e}")
423 raise
425 def resume_subscription(self, subscription_id: str) -> dict[str, str]:
426 """Resume a paused subscription."""
427 try:
428 stripe.Subscription.modify(
429 subscription_id,
430 pause_collection="", # Remove pause
431 )
432 return {"message": "Subscription resumed", "status": "active"}
433 except stripe.error.StripeError as e:
434 logger.error(f"Stripe error resuming subscription: {e}")
435 raise
437 def cancel_subscription(self, subscription_id: str) -> dict[str, str]:
438 """Cancel a subscription."""
439 try:
440 stripe.Subscription.cancel(subscription_id)
441 return {"message": "Subscription cancelled", "status": "cancelled"}
442 except stripe.error.StripeError as e:
443 logger.error(f"Stripe error cancelling subscription: {e}")
444 raise
446 def create_billing_portal_session(self, customer_id: str, return_url: str) -> str:
447 """Create a Stripe Billing Portal session for the customer.
449 Returns the portal session URL.
450 """
451 try:
452 session = stripe.billing_portal.Session.create(
453 customer=customer_id,
454 return_url=return_url,
455 )
456 return session.url
457 except stripe.error.StripeError as e:
458 logger.error(f"Stripe error creating billing portal session: {e}")
459 raise