Coverage for polar/customer_portal/endpoints/customer.py: 89%
45 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from fastapi import Depends, Request
2from pydantic import UUID4
3from sse_starlette import EventSourceResponse
5from polar.eventstream.endpoints import subscribe
6from polar.eventstream.service import Receivers
7from polar.exceptions import ResourceNotFound
8from polar.kit.pagination import ListResource, PaginationParamsQuery
9from polar.models import Customer
10from polar.openapi import APITag
11from polar.payment_method.service import PaymentMethodInUseByActiveSubscription
12from polar.postgres import AsyncSession, get_db_session
13from polar.redis import Redis, get_redis
14from polar.routing import APIRouter
16from .. import auth
17from ..schemas.customer import (
18 CustomerPaymentMethod,
19 CustomerPaymentMethodConfirm,
20 CustomerPaymentMethodCreate,
21 CustomerPaymentMethodCreateResponse,
22 CustomerPaymentMethodTypeAdapter,
23 CustomerPortalCustomer,
24 CustomerPortalCustomerUpdate,
25)
26from ..service.customer import CustomerNotReady
27from ..service.customer import customer as customer_service
29router = APIRouter(prefix="/customers", tags=["customers", APITag.public])
32@router.get("/stream", include_in_schema=False)
33async def stream(
34 request: Request,
35 auth_subject: auth.CustomerPortalRead,
36 session: AsyncSession = Depends(get_db_session),
37 redis: Redis = Depends(get_redis),
38) -> EventSourceResponse:
39 receivers = Receivers(customer_id=auth_subject.subject.id)
40 channels = receivers.get_channels()
41 return EventSourceResponse(subscribe(redis, channels, request))
44@router.get("/me", summary="Get Customer", response_model=CustomerPortalCustomer)
45async def get(auth_subject: auth.CustomerPortalRead) -> Customer:
46 """Get authenticated customer."""
47 return auth_subject.subject
50@router.patch(
51 "/me",
52 summary="Update Customer",
53 responses={
54 200: {"description": "Customer updated."},
55 },
56 response_model=CustomerPortalCustomer,
57)
58async def update(
59 customer_update: CustomerPortalCustomerUpdate,
60 auth_subject: auth.CustomerPortalWrite,
61 session: AsyncSession = Depends(get_db_session),
62) -> Customer:
63 """Update authenticated customer."""
64 return await customer_service.update(session, auth_subject.subject, customer_update)
67@router.get(
68 "/me/payment-methods",
69 summary="List Customer Payment Methods",
70 response_model=ListResource[CustomerPaymentMethod],
71)
72async def list_payment_methods(
73 auth_subject: auth.CustomerPortalRead,
74 pagination: PaginationParamsQuery,
75 session: AsyncSession = Depends(get_db_session),
76) -> ListResource[CustomerPaymentMethod]:
77 """Get saved payment methods of the authenticated customer."""
78 results, count = await customer_service.list_payment_methods(
79 session, auth_subject, pagination=pagination
80 )
81 return ListResource.from_paginated_results(
82 [
83 CustomerPaymentMethodTypeAdapter.validate_python(result)
84 for result in results
85 ],
86 count,
87 pagination,
88 )
91@router.post(
92 "/me/payment-methods",
93 summary="Add Customer Payment Method",
94 status_code=201,
95 responses={
96 201: {"description": "Payment method created or setup initiated."},
97 },
98 response_model=CustomerPaymentMethodCreateResponse,
99)
100async def add_payment_method(
101 auth_subject: auth.CustomerPortalRead,
102 payment_method_create: CustomerPaymentMethodCreate,
103 session: AsyncSession = Depends(get_db_session),
104) -> CustomerPaymentMethodCreateResponse:
105 """Add a payment method to the authenticated customer."""
106 return await customer_service.add_payment_method(
107 session, auth_subject.subject, payment_method_create
108 )
111@router.post(
112 "/me/payment-methods/confirm",
113 summary="Confirm Customer Payment Method",
114 status_code=201,
115 responses={
116 201: {"description": "Payment method created or setup initiated."},
117 400: {
118 "description": "Customer is not ready to confirm a payment method.",
119 "model": CustomerNotReady.schema(),
120 },
121 },
122 response_model=CustomerPaymentMethodCreateResponse,
123)
124async def confirm_payment_method(
125 auth_subject: auth.CustomerPortalRead,
126 payment_method_confirm: CustomerPaymentMethodConfirm,
127 session: AsyncSession = Depends(get_db_session),
128) -> CustomerPaymentMethodCreateResponse:
129 """Confirm a payment method for the authenticated customer."""
130 return await customer_service.confirm_payment_method(
131 session, auth_subject.subject, payment_method_confirm
132 )
135@router.delete(
136 "/me/payment-methods/{id}",
137 summary="Delete Customer Payment Method",
138 status_code=204,
139 responses={
140 204: {"description": "Payment method deleted."},
141 400: {
142 "description": "Payment method is used by active subscription(s).",
143 "model": PaymentMethodInUseByActiveSubscription.schema(),
144 },
145 404: {
146 "description": "Payment method not found.",
147 "model": ResourceNotFound.schema(),
148 },
149 },
150)
151async def delete_payment_method(
152 id: UUID4,
153 auth_subject: auth.CustomerPortalRead,
154 session: AsyncSession = Depends(get_db_session),
155) -> None:
156 """Delete a payment method from the authenticated customer."""
157 payment_method = await customer_service.get_payment_method(
158 session, auth_subject, id
159 )
160 if payment_method is None: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true
161 raise ResourceNotFound()
162 await customer_service.delete_payment_method(session, payment_method)