Coverage for src/couchers/servicers/donations.py: 95%

1import json 

2import logging 

3 

4import grpc 

5import stripe 

6 

7from couchers import errors, urls 

8from couchers.config import config 

9from couchers.models import DonationInitiation, DonationType, Invoice, User 

10from couchers.notifications.notify import notify 

11from couchers.sql import couchers_select as select 

12from proto import donations_pb2, donations_pb2_grpc, notification_data_pb2, stripe_pb2_grpc 

13from proto.google.api import httpbody_pb2 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18def _create_stripe_customer(session, user): 

19 # create a new stripe id for this user 

20 customer = stripe.Customer.create( 

21 email=user.email, 

22 # metadata allows us to store arbitrary metadata for ourselves 

23 metadata={"user_id": user.id}, 

24 api_key=config["STRIPE_API_KEY"], 

25 ) 

26 user.stripe_customer_id = customer.id 

27 # commit since we only ever want one stripe customer id per user, so if the rest of this api call fails, this will still be saved in the db 

28 session.commit() 

29 

30 

31class Donations(donations_pb2_grpc.DonationsServicer): 

32 def InitiateDonation(self, request, context, session): 

33 if not config["ENABLE_DONATIONS"]: 

34 context.abort(grpc.StatusCode.UNAVAILABLE, errors.DONATIONS_DISABLED) 

35 

36 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

37 

38 if request.amount < 2: 

39 # we don't want to waste *all* of the donation on processing fees 

40 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.DONATION_TOO_SMALL) 

41 

42 if not user.stripe_customer_id: 

43 _create_stripe_customer(session, user) 

44 

45 if request.recurring: 

46 item = { 

47 "price": config["STRIPE_RECURRING_PRODUCT_ID"], 

48 "quantity": request.amount, 

49 } 

50 else: 

51 item = { 

52 "price_data": { 

53 "currency": "usd", 

54 "unit_amount": request.amount * 100, # input is in cents 

55 "product_data": { 

56 "name": "Couchers financial supporter (one-time)", 

57 "images": ["https://couchers.org/img/share.jpg"], 

58 }, 

59 }, 

60 "quantity": 1, 

61 } 

62 

63 checkout_session = stripe.checkout.Session.create( 

64 client_reference_id=user.id, 

65 submit_type="donate" if not request.recurring else None, 

66 customer=user.stripe_customer_id, 

67 success_url=urls.donation_success_url(), 

68 cancel_url=urls.donation_cancelled_url(), 

69 payment_method_types=["card"], 

70 mode="subscription" if request.recurring else "payment", 

71 line_items=[item], 

72 api_key=config["STRIPE_API_KEY"], 

73 ) 

74 

75 session.add( 

76 DonationInitiation( 

77 user_id=user.id, 

78 amount=request.amount, 

79 stripe_checkout_session_id=checkout_session.id, 

80 donation_type=DonationType.recurring if request.recurring else DonationType.one_time, 

81 ) 

82 ) 

83 

84 return donations_pb2.InitiateDonationRes( 

85 stripe_checkout_session_id=checkout_session.id, stripe_checkout_url=checkout_session.url 

86 ) 

87 

88 def GetDonationPortalLink(self, request, context, session): 

89 if not config["ENABLE_DONATIONS"]: 

90 context.abort(grpc.StatusCode.UNAVAILABLE, errors.DONATIONS_DISABLED) 

91 

92 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

93 

94 if not user.stripe_customer_id: 

95 _create_stripe_customer(session, user) 

96 

97 session = stripe.billing_portal.Session.create( 

98 customer=user.stripe_customer_id, 

99 return_url=urls.donation_url(), 

100 api_key=config["STRIPE_API_KEY"], 

101 ) 

102 

103 return donations_pb2.GetDonationPortalLinkRes(stripe_portal_url=session.url) 

104 

105 

106class Stripe(stripe_pb2_grpc.StripeServicer): 

107 def Webhook(self, request, context, session): 

108 # We're set up to receive the following webhook events (with explanations from stripe docs): 

109 # For both recurring and one-off donations, we get a `charge.succeeded` event and we then send the user an 

110 # invoice. There are other events too, but we don't handle them right now. 

111 headers = dict(context.invocation_metadata()) 

112 

113 event = stripe.Webhook.construct_event( 

114 payload=request.data, 

115 sig_header=headers.get("stripe-signature"), 

116 secret=config["STRIPE_WEBHOOK_SECRET"], 

117 api_key=config["STRIPE_API_KEY"], 

118 ) 

119 data = event["data"] 

120 event_type = event["type"] 

121 event_id = event["id"] 

122 data_object = data["object"] 

123 

124 # Get the type of webhook event sent - used to check the status of PaymentIntents. 

125 logger.info(f"Got signed Stripe webhook, {event_type=}, {event_id=}") 

126 

127 if event_type == "charge.succeeded": 

128 customer_id = data_object["customer"] 

129 user = session.execute(select(User).where(User.stripe_customer_id == customer_id)).scalar_one() 

130 # amount comes in cents 

131 amount = int(data_object["amount"]) // 100 

132 receipt_url = data_object["receipt_url"] 

133 

134 # may be check for amount to enable phone verify 

135 user.has_donated = True 

136 

137 session.add( 

138 Invoice( 

139 user_id=user.id, 

140 amount=amount, 

141 stripe_payment_intent_id=data_object["payment_intent"], 

142 stripe_receipt_url=receipt_url, 

143 ) 

144 ) 

145 

146 notify( 

147 session, 

148 user_id=user.id, 

149 topic_action="donation:received", 

150 data=notification_data_pb2.DonationReceived( 

151 amount=amount, 

152 receipt_url=receipt_url, 

153 ), 

154 ) 

155 else: 

156 logger.info(f"Unhandled event from Stripe: {event_type}") 

157 

158 return httpbody_pb2.HttpBody( 

159 content_type="application/json", 

160 # json.dumps escapes non-ascii characters 

161 data=json.dumps({"success": True}).encode("ascii"), 

162 )