Coverage for backend / app / payments / customer.py: 100%

1"""Stripe Customer API""" 

2 

3import time 

4 

5from fastapi import HTTPException 

6from sqlalchemy.exc import SQLAlchemyError 

7from sqlalchemy.orm import Session 

8 

9from app.config import settings 

10from app.models import User 

11from app.payments import stripe, logger 

12 

13 

14async def create_customer(user: User) -> stripe.Customer: 

15 """Create a Stripe customer, with test clock if in test mode. 

16 :param user: User object 

17 :return: Customer object""" 

18 

19 customer_params = {"email": user.email, "metadata": {"user_id": str(user.id)}} 

20 

21 if settings.test_mode: 

22 test_clock = await stripe.test_helpers.TestClock.create_async( 

23 frozen_time=int(time.time()), 

24 name=f"Test clock for user {user.id}", 

25 ) 

26 customer_params["test_clock"] = test_clock.id 

27 logger.info(f"Created test clock {test_clock.id} for user {user.id}") 

28 

29 return await stripe.Customer.create_async(**customer_params) 

30 

31 

32async def get_or_create_stripe_customer( 

33 user: User, 

34 db: Session, 

35) -> str: 

36 """Get or create Stripe customer for the user. 

37 :param user: User object 

38 :param db: Database session 

39 :return: Customer ID string 

40 :raises HTTPException: On validation or Stripe errors""" 

41 

42 try: 

43 # If the user as a customer id, retrieve the existing customer 

44 if user.stripe_details.customer_id: 

45 customer = await stripe.Customer.retrieve_async(user.stripe_details.customer_id) 

46 

47 # If a non-deleted Stripe customer was found 

48 if customer and not customer.get("deleted", False): # noqa 

49 # Update Stripe email if it doesn't match our database 

50 if customer["email"] != user.email: 

51 customer = await stripe.Customer.modify_async(user.stripe_details.customer_id, email=user.email) 

52 logger.info(f"Updated Stripe customer {customer.id} email to {user.email}") 

53 else: 

54 logger.info(f"Retrieved existing Stripe customer {customer.id} for user {user.id}") 

55 else: 

56 customer = await create_customer(user) 

57 user.stripe_details.customer_id = customer.id 

58 db.commit() 

59 logger.info(f"Created new Stripe customer {customer.id} for user {user.id}") 

60 

61 # If no customer, create one 

62 else: 

63 customer = await create_customer(user) 

64 user.stripe_details.customer_id = customer.id 

65 db.commit() 

66 logger.info(f"Created new Stripe customer {customer.id} for user {user.id}") 

67 

68 return user.stripe_details.customer_id 

69 

70 except stripe.error.StripeError as e: 

71 logger.error(f"Stripe API error for user {user.id}: {str(e)}", exc_info=True) 

72 db.rollback() 

73 raise HTTPException(status_code=503, detail="Payment service temporarily unavailable. Please try again.") 

74 except SQLAlchemyError as e: 

75 logger.error(f"Database error for user {user.id}: {str(e)}", exc_info=True) 

76 db.rollback() 

77 raise HTTPException(status_code=500, detail="An error occurred. Please try again.")