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

1import json 

2import logging 

3 

4import grpc 

5import stripe 

6 

7from couchers import errors, urls 

8from couchers.config import config 

9from couchers.db import session_scope 

10from couchers.models import Invoice, OneTimeDonation, RecurringDonation, User 

11from couchers.sql import couchers_select as select 

12from couchers.tasks import send_donation_email 

13from couchers.utils import now 

14from proto import donations_pb2, donations_pb2_grpc, stripe_pb2_grpc 

15from proto.google.api import httpbody_pb2 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class Donations(donations_pb2_grpc.DonationsServicer): 

21 def InitiateDonation(self, request, context): 

22 if not config["ENABLE_DONATIONS"]: 

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

24 

25 with session_scope() as session: 

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

27 

28 if request.amount < 2: 

29 # we don't want to waste *all* the donations on processing fees 

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

31 

32 if not user.stripe_customer_id: 

33 # create a new stripe id for this user 

34 customer = stripe.Customer.create( 

35 email=user.email, 

36 # metadata allows us to store arbitrary metadata for ourselves 

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

38 api_key=config["STRIPE_API_KEY"], 

39 ) 

40 user.stripe_customer_id = customer.id 

41 # 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 

42 session.commit() 

43 

44 if request.recurring: 

45 item = { 

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

47 "quantity": request.amount, 

48 } 

49 else: 

50 item = { 

51 "price_data": { 

52 "currency": "usd", 

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

54 "product_data": { 

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

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

57 }, 

58 }, 

59 "quantity": 1, 

60 } 

61 

62 checkout_session = stripe.checkout.Session.create( 

63 client_reference_id=user.id, 

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

65 customer=user.stripe_customer_id, 

66 success_url=urls.donation_success_url(), 

67 cancel_url=urls.donation_cancelled_url(), 

68 payment_method_types=["card"], 

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

70 line_items=[item], 

71 api_key=config["STRIPE_API_KEY"], 

72 ) 

73 

74 if request.recurring: 

75 session.add( 

76 RecurringDonation( 

77 user_id=user.id, 

78 amount=request.amount, 

79 stripe_checkout_session_id=checkout_session.id, 

80 ) 

81 ) 

82 else: 

83 session.add( 

84 OneTimeDonation( 

85 user_id=user.id, 

86 amount=request.amount, 

87 stripe_checkout_session_id=checkout_session.id, 

88 stripe_payment_intent_id=checkout_session.payment_intent, 

89 paid=None, 

90 ) 

91 ) 

92 

93 return donations_pb2.InitiateDonationRes(stripe_checkout_session_id=checkout_session.id) 

94 

95 

96class Stripe(stripe_pb2_grpc.StripeServicer): 

97 def Webhook(self, request, context): 

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

99 # For both recurring and one-off donations, we get a `checkout.session.completed` event, and then a `payment_intent.succeeded` event 

100 headers = dict(context.invocation_metadata()) 

101 

102 event = stripe.Webhook.construct_event( 

103 payload=request.data, 

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

105 secret=config["STRIPE_WEBHOOK_SECRET"], 

106 api_key=config["STRIPE_API_KEY"], 

107 ) 

108 data = event["data"] 

109 event_type = event["type"] 

110 event_id = event["id"] 

111 data_object = data["object"] 

112 

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

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

115 

116 if event_type == "checkout.session.completed": 

117 checkout_session_id = data_object["id"] 

118 if data_object["payment_intent"]: 

119 with session_scope() as session: 

120 donation = session.execute( 

121 select(OneTimeDonation).where(OneTimeDonation.stripe_checkout_session_id == checkout_session_id) 

122 ).scalar_one() 

123 if data_object["payment_status"] == "paid": 

124 donation.paid = now() 

125 else: 

126 raise Exception("Unknown payment status") 

127 elif data_object["subscription"]: 

128 with session_scope() as session: 

129 donation = session.execute( 

130 select(RecurringDonation).where( 

131 RecurringDonation.stripe_checkout_session_id == checkout_session_id 

132 ) 

133 ).scalar_one() 

134 donation.stripe_subscription_id = data_object["subscription"] 

135 else: 

136 raise Exception("Unknown payment type") 

137 elif event_type == "payment_intent.succeeded": 

138 customer_id = data_object["customer"] 

139 with session_scope() as session: 

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

141 invoice_data = data_object["charges"]["data"][0] 

142 # amount comes in cents 

143 amount = int(float(invoice_data["amount"]) / 100) 

144 receipt_url = invoice_data["receipt_url"] 

145 session.add( 

146 Invoice( 

147 user_id=user.id, 

148 amount=amount, 

149 stripe_payment_intent_id=invoice_data["payment_intent"], 

150 stripe_receipt_url=receipt_url, 

151 ) 

152 ) 

153 send_donation_email(user, amount, receipt_url) 

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 )