Coverage for polar/payment_method/repository.py: 45%

1from uuid import UUID 1 ctx1a

2 

3from sqlalchemy import update 1 ctx1a

4from sqlalchemy.orm import joinedload 1 ctx1a

5 

6from polar.enums import PaymentProcessor 1 ctx1a

7from polar.kit.repository import ( 1 ctx1a

8 Options, 

9 RepositoryBase, 

10 RepositorySoftDeletionIDMixin, 

11 RepositorySoftDeletionMixin, 

12) 

13from polar.models import Customer, PaymentMethod, Subscription 1 ctx1a

14 

15 

16class PaymentMethodRepository( 1 ctx1a

17 RepositorySoftDeletionIDMixin[PaymentMethod, UUID], 

18 RepositorySoftDeletionMixin[PaymentMethod], 

19 RepositoryBase[PaymentMethod], 

20): 

21 model = PaymentMethod 1 ctx1a

22 

23 async def get_by_id_and_customer( 1 ctx1a

24 self, 

25 id: UUID, 

26 customer: UUID, 

27 *, 

28 options: Options = (), 

29 ) -> PaymentMethod | None: 

30 statement = ( 

31 self.get_base_statement() 

32 .where( 

33 PaymentMethod.id == id, 

34 PaymentMethod.customer_id == customer, 

35 ) 

36 .options(*options) 

37 ) 

38 return await self.get_one_or_none(statement) 

39 

40 async def get_by_customer_and_processor_id( 1 ctx1a

41 self, 

42 customer: UUID, 

43 processor: PaymentProcessor, 

44 processor_id: str, 

45 *, 

46 options: Options = (), 

47 include_deleted: bool = False, 

48 ) -> PaymentMethod | None: 

49 statement = ( 

50 self.get_base_statement(include_deleted=include_deleted) 

51 .where( 

52 PaymentMethod.customer_id == customer, 

53 PaymentMethod.processor == processor, 

54 PaymentMethod.processor_id == processor_id, 

55 ) 

56 .options(*options) 

57 ) 

58 return await self.get_one_or_none(statement) 

59 

60 async def list_by_customer( 1 ctx1a

61 self, 

62 customer_id: UUID, 

63 *, 

64 exclude_id: UUID | None = None, 

65 options: Options = (), 

66 ) -> list[PaymentMethod]: 

67 statement = self.get_base_statement().where( 

68 PaymentMethod.customer_id == customer_id 

69 ) 

70 if exclude_id is not None: 

71 statement = statement.where(PaymentMethod.id != exclude_id) 

72 statement = statement.options(*options) 

73 result = await self.session.execute(statement) 

74 return list(result.scalars().all()) 

75 

76 async def soft_delete( 1 ctx1a

77 self, object: PaymentMethod, *, flush: bool = False 

78 ) -> PaymentMethod: 

79 # Unlink the payment method from the customer and subscriptions 

80 await self.session.execute( 

81 update(Customer) 

82 .values(default_payment_method_id=None) 

83 .where(Customer.default_payment_method_id == object.id) 

84 ) 

85 await self.session.execute( 

86 update(Subscription) 

87 .values(payment_method_id=None) 

88 .where(Subscription.payment_method_id == object.id) 

89 ) 

90 

91 return await super().soft_delete(object, flush=flush) 

92 

93 def get_eager_options(self) -> Options: 1 ctx1a

94 return (joinedload(PaymentMethod.customer),)