Coverage for app / backend / src / couchers / servicers / donations.py: 92%

1import json 

2import logging 

3 

4import grpc 

5import stripe 

6from google.protobuf import empty_pb2 

7from sqlalchemy import select 

8from sqlalchemy.orm import Session 

9 

10from couchers import urls 

11from couchers.config import config 

12from couchers.context import CouchersContext 

13from couchers.helpers.badges import user_add_badge 

14from couchers.models import DonationInitiation, DonationType, Invoice, InvoiceType, User 

15from couchers.models.notifications import NotificationTopicAction 

16from couchers.notifications.notify import notify 

17from couchers.proto import donations_pb2, donations_pb2_grpc, notification_data_pb2, stripe_pb2_grpc 

18from couchers.proto.google.api import httpbody_pb2 

19from couchers.utils import not_none 

20 

21logger = logging.getLogger(__name__) 

22 

23 

24def _create_stripe_customer(session: Session, user: User) -> None: 

25 # create a new stripe id for this user 

26 customer = stripe.Customer.create( 

27 email=user.email, 

28 # metadata allows us to store arbitrary metadata for ourselves 

29 metadata={"user_id": user.id}, # type: ignore[dict-item] 

30 api_key=config["STRIPE_API_KEY"], 

31 ) 

32 user.stripe_customer_id = customer.id 

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

34 session.commit() 

35 

36 

37class Donations(donations_pb2_grpc.DonationsServicer): 

38 def InitiateDonation( 

39 self, request: donations_pb2.InitiateDonationReq, context: CouchersContext, session: Session 

40 ) -> donations_pb2.InitiateDonationRes: 

41 if not config["ENABLE_DONATIONS"]: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "donations_disabled") 

43 

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

45 

46 if request.amount < 2: 46 ↛ 48line 46 didn't jump to line 48 because the condition on line 46 was never true

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

48 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "donation_too_small") 

49 

50 if not user.stripe_customer_id: 50 ↛ 53line 50 didn't jump to line 53 because the condition on line 50 was always true

51 _create_stripe_customer(session, user) 

52 

53 if request.recurring: 

54 item = { 

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

56 "quantity": request.amount, 

57 } 

58 else: 

59 item = { 

60 "price_data": { 

61 "currency": "usd", 

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

63 "product_data": { 

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

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

66 }, 

67 }, 

68 "quantity": 1, 

69 } 

70 

71 checkout_session = stripe.checkout.Session.create( 

72 client_reference_id=str(user.id), 

73 # Stripe actually allows None, but the signature says it's either a string or not passed. 

74 submit_type="donate" if not request.recurring else None, # type: ignore[arg-type] 

75 customer=not_none(user.stripe_customer_id), 

76 success_url=urls.donation_success_url(), 

77 cancel_url=urls.donation_cancelled_url(), 

78 payment_method_types=["card"], 

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

80 line_items=[item], # type: ignore[list-item] 

81 api_key=config["STRIPE_API_KEY"], 

82 ) 

83 

84 session.add( 

85 DonationInitiation( 

86 user_id=user.id, 

87 amount=request.amount, 

88 stripe_checkout_session_id=checkout_session.id, 

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

90 source=request.source if request.source else None, 

91 ) 

92 ) 

93 

94 return donations_pb2.InitiateDonationRes( 

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

96 ) 

97 

98 def GetDonationPortalLink( 

99 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

100 ) -> donations_pb2.GetDonationPortalLinkRes: 

101 if not config["ENABLE_DONATIONS"]: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "donations_disabled") 

103 

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

105 

106 if not user.stripe_customer_id: 

107 _create_stripe_customer(session, user) 

108 

109 stripe_session = stripe.billing_portal.Session.create( 

110 customer=not_none(user.stripe_customer_id), 

111 return_url=urls.donation_url(), 

112 api_key=config["STRIPE_API_KEY"], 

113 ) 

114 

115 return donations_pb2.GetDonationPortalLinkRes(stripe_portal_url=stripe_session.url) 

116 

117 

118class Stripe(stripe_pb2_grpc.StripeServicer): 

119 def Webhook( 

120 self, request: httpbody_pb2.HttpBody, context: CouchersContext, session: Session 

121 ) -> httpbody_pb2.HttpBody: 

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

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

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

125 event = stripe.Webhook.construct_event( # type: ignore[no-untyped-call] 

126 payload=request.data, 

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

128 secret=config["STRIPE_WEBHOOK_SECRET"], 

129 api_key=config["STRIPE_API_KEY"], 

130 ) 

131 data = event["data"] 

132 event_type = event["type"] 

133 event_id = event["id"] 

134 data_object = data["object"] 

135 metadata = data_object.get("metadata", {}) 

136 

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

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

139 

140 if event_type == "charge.succeeded": 

141 if metadata.get("site_url") == config["MERCH_SHOP_URL"]: 

142 # merch shop. look up this email and give them the swagster badge 

143 user = session.execute( 

144 select(User).where(User.email == metadata["customer_email"]) 

145 ).scalar_one_or_none() 

146 if user: 

147 user_add_badge(session, user.id, "swagster") 

148 else: 

149 customer_id = data_object["customer"] 

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

151 # amount comes in cents 

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

153 receipt_url = data_object["receipt_url"] 

154 payment_intent_id = data_object["payment_intent"] 

155 

156 invoice = Invoice( 

157 user_id=user.id, 

158 amount=amount, 

159 stripe_payment_intent_id=payment_intent_id, 

160 stripe_receipt_url=receipt_url, 

161 invoice_type=InvoiceType.on_platform, 

162 ) 

163 session.add(invoice) 

164 session.flush() 

165 user.last_donated = invoice.created 

166 

167 notify( 

168 session, 

169 user_id=user.id, 

170 topic_action=NotificationTopicAction.donation__received, 

171 key="", 

172 data=notification_data_pb2.DonationReceived( 

173 amount=amount, 

174 receipt_url=receipt_url, 

175 ), 

176 ) 

177 else: 

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

179 

180 return httpbody_pb2.HttpBody( 

181 content_type="application/json", 

182 # json.dumps escapes non-ascii characters 

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

184 )