Coverage for src/couchers/servicers/donations.py: 95%
66 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-14 00:52 +0000
1import json
2import logging
4import grpc
5import stripe
7from couchers import urls
8from couchers.config import config
9from couchers.helpers.badges import user_add_badge
10from couchers.models import DonationInitiation, DonationType, Invoice, InvoiceType, User
11from couchers.notifications.notify import notify
12from couchers.proto import donations_pb2, donations_pb2_grpc, notification_data_pb2, stripe_pb2_grpc
13from couchers.proto.google.api import httpbody_pb2
14from couchers.sql import couchers_select as select
16logger = logging.getLogger(__name__)
19def _create_stripe_customer(session, user):
20 # create a new stripe id for this user
21 customer = stripe.Customer.create(
22 email=user.email,
23 # metadata allows us to store arbitrary metadata for ourselves
24 metadata={"user_id": user.id},
25 api_key=config["STRIPE_API_KEY"],
26 )
27 user.stripe_customer_id = customer.id
28 # 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
29 session.commit()
32class Donations(donations_pb2_grpc.DonationsServicer):
33 def InitiateDonation(self, request, context, session):
34 if not config["ENABLE_DONATIONS"]:
35 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "donations_disabled")
37 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
39 if request.amount < 2:
40 # we don't want to waste *all* of the donation on processing fees
41 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "donation_too_small")
43 if not user.stripe_customer_id:
44 _create_stripe_customer(session, user)
46 if request.recurring:
47 item = {
48 "price": config["STRIPE_RECURRING_PRODUCT_ID"],
49 "quantity": request.amount,
50 }
51 else:
52 item = {
53 "price_data": {
54 "currency": "usd",
55 "unit_amount": request.amount * 100, # input is in cents
56 "product_data": {
57 "name": "Couchers financial supporter (one-time)",
58 "images": ["https://couchers.org/img/share.jpg"],
59 },
60 },
61 "quantity": 1,
62 }
64 checkout_session = stripe.checkout.Session.create(
65 client_reference_id=user.id,
66 submit_type="donate" if not request.recurring else None,
67 customer=user.stripe_customer_id,
68 success_url=urls.donation_success_url(),
69 cancel_url=urls.donation_cancelled_url(),
70 payment_method_types=["card"],
71 mode="subscription" if request.recurring else "payment",
72 line_items=[item],
73 api_key=config["STRIPE_API_KEY"],
74 )
76 session.add(
77 DonationInitiation(
78 user_id=user.id,
79 amount=request.amount,
80 stripe_checkout_session_id=checkout_session.id,
81 donation_type=DonationType.recurring if request.recurring else DonationType.one_time,
82 source=request.source if request.source else None,
83 )
84 )
86 return donations_pb2.InitiateDonationRes(
87 stripe_checkout_session_id=checkout_session.id, stripe_checkout_url=checkout_session.url
88 )
90 def GetDonationPortalLink(self, request, context, session):
91 if not config["ENABLE_DONATIONS"]:
92 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "donations_disabled")
94 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
96 if not user.stripe_customer_id:
97 _create_stripe_customer(session, user)
99 session = stripe.billing_portal.Session.create(
100 customer=user.stripe_customer_id,
101 return_url=urls.donation_url(),
102 api_key=config["STRIPE_API_KEY"],
103 )
105 return donations_pb2.GetDonationPortalLinkRes(stripe_portal_url=session.url)
108class Stripe(stripe_pb2_grpc.StripeServicer):
109 def Webhook(self, request, context, session):
110 # We're set up to receive the following webhook events (with explanations from stripe docs):
111 # For both recurring and one-off donations, we get a `charge.succeeded` event and we then send the user an
112 # invoice. There are other events too, but we don't handle them right now.
113 event = stripe.Webhook.construct_event(
114 payload=request.data,
115 sig_header=context.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 metadata = data_object.get("metadata", {})
125 # Get the type of webhook event sent - used to check the status of PaymentIntents.
126 logger.info(f"Got signed Stripe webhook, {event_type=}, {event_id=}")
128 if event_type == "charge.succeeded":
129 if metadata.get("site_url") == config["MERCH_SHOP_URL"]:
130 # merch shop. look up this email and give them the swagster badge
131 user = session.execute(
132 select(User).where(User.email == metadata["customer_email"])
133 ).scalar_one_or_none()
134 if user:
135 user_add_badge(session, user.id, "swagster")
136 else:
137 customer_id = data_object["customer"]
138 user = session.execute(select(User).where(User.stripe_customer_id == customer_id)).scalar_one()
139 # amount comes in cents
140 amount = int(data_object["amount"]) // 100
141 receipt_url = data_object["receipt_url"]
142 payment_intent_id = data_object["payment_intent"]
144 invoice = Invoice(
145 user_id=user.id,
146 amount=amount,
147 stripe_payment_intent_id=payment_intent_id,
148 stripe_receipt_url=receipt_url,
149 invoice_type=InvoiceType.on_platform,
150 )
151 session.add(invoice)
152 session.flush()
153 user.last_donated = invoice.created
155 notify(
156 session,
157 user_id=user.id,
158 topic_action="donation:received",
159 key="",
160 data=notification_data_pb2.DonationReceived(
161 amount=amount,
162 receipt_url=receipt_url,
163 ),
164 )
165 else:
166 logger.info(f"Unhandled event from Stripe: {event_type}")
168 return httpbody_pb2.HttpBody(
169 content_type="application/json",
170 # json.dumps escapes non-ascii characters
171 data=json.dumps({"success": True}).encode("ascii"),
172 )