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 source=request.source if request.source else None, 

82 ) 

83 ) 

84 

85 return donations_pb2.InitiateDonationRes( 

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

87 ) 

88 

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

90 if not config["ENABLE_DONATIONS"]: 

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

92 

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

94 

95 if not user.stripe_customer_id: 

96 _create_stripe_customer(session, user) 

97 

98 session = stripe.billing_portal.Session.create( 

99 customer=user.stripe_customer_id, 

100 return_url=urls.donation_url(), 

101 api_key=config["STRIPE_API_KEY"], 

102 ) 

103 

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

105 

106 

107class Stripe(stripe_pb2_grpc.StripeServicer): 

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

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

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

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

112 event = stripe.Webhook.construct_event( 

113 payload=request.data, 

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

115 secret=config["STRIPE_WEBHOOK_SECRET"], 

116 api_key=config["STRIPE_API_KEY"], 

117 ) 

118 data = event["data"] 

119 event_type = event["type"] 

120 event_id = event["id"] 

121 data_object = data["object"] 

122 

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

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

125 

126 if event_type == "charge.succeeded": 

127 customer_id = data_object["customer"] 

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

129 # amount comes in cents 

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

131 receipt_url = data_object["receipt_url"] 

132 

133 # may be check for amount to enable phone verify 

134 user.has_donated = True 

135 

136 session.add( 

137 Invoice( 

138 user_id=user.id, 

139 amount=amount, 

140 stripe_payment_intent_id=data_object["payment_intent"], 

141 stripe_receipt_url=receipt_url, 

142 ) 

143 ) 

144 

145 notify( 

146 session, 

147 user_id=user.id, 

148 topic_action="donation:received", 

149 data=notification_data_pb2.DonationReceived( 

150 amount=amount, 

151 receipt_url=receipt_url, 

152 ), 

153 ) 

154 else: 

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

156 

157 return httpbody_pb2.HttpBody( 

158 content_type="application/json", 

159 # json.dumps escapes non-ascii characters 

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

161 )