Coverage for src/couchers/servicers/donations.py: 95%
58 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-07-09 00:05 +0000
« prev ^ index » next coverage.py v7.6.10, created at 2025-07-09 00:05 +0000
1import json
2import logging
4import grpc
5import stripe
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
15logger = logging.getLogger(__name__)
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()
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)
36 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
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)
42 if not user.stripe_customer_id:
43 _create_stripe_customer(session, user)
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 }
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 )
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 )
84 return donations_pb2.InitiateDonationRes(
85 stripe_checkout_session_id=checkout_session.id, stripe_checkout_url=checkout_session.url
86 )
88 def GetDonationPortalLink(self, request, context, session):
89 if not config["ENABLE_DONATIONS"]:
90 context.abort(grpc.StatusCode.UNAVAILABLE, errors.DONATIONS_DISABLED)
92 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
94 if not user.stripe_customer_id:
95 _create_stripe_customer(session, user)
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 )
103 return donations_pb2.GetDonationPortalLinkRes(stripe_portal_url=session.url)
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())
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"]
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=}")
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"]
134 # may be check for amount to enable phone verify
135 user.has_donated = True
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 )
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}")
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 )