Coverage for polar/customer_portal/endpoints/customer.py: 68%
45 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends, Request 1a
2from pydantic import UUID4 1a
3from sse_starlette import EventSourceResponse 1a
5from polar.eventstream.endpoints import subscribe 1a
6from polar.eventstream.service import Receivers 1a
7from polar.exceptions import ResourceNotFound 1a
8from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
9from polar.models import Customer 1a
10from polar.openapi import APITag 1a
11from polar.payment_method.service import PaymentMethodInUseByActiveSubscription 1a
12from polar.postgres import AsyncSession, get_db_session 1a
13from polar.redis import Redis, get_redis 1a
14from polar.routing import APIRouter 1a
16from .. import auth 1a
17from ..schemas.customer import ( 1a
18 CustomerPaymentMethod,
19 CustomerPaymentMethodConfirm,
20 CustomerPaymentMethodCreate,
21 CustomerPaymentMethodCreateResponse,
22 CustomerPaymentMethodTypeAdapter,
23 CustomerPortalCustomer,
24 CustomerPortalCustomerUpdate,
25)
26from ..service.customer import CustomerNotReady 1a
27from ..service.customer import customer as customer_service 1a
29router = APIRouter(prefix="/customers", tags=["customers", APITag.public]) 1a
32@router.get("/stream", include_in_schema=False) 1a
33async def stream( 1a
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) 1a
45async def get(auth_subject: auth.CustomerPortalRead) -> Customer: 1a
46 """Get authenticated customer."""
47 return auth_subject.subject
50@router.patch( 1a
51 "/me",
52 summary="Update Customer",
53 responses={
54 200: {"description": "Customer updated."},
55 },
56 response_model=CustomerPortalCustomer,
57)
58async def update( 1a
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( 1a
68 "/me/payment-methods",
69 summary="List Customer Payment Methods",
70 response_model=ListResource[CustomerPaymentMethod],
71)
72async def list_payment_methods( 1a
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( 1a
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( 1a
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( 1a
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( 1a
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( 1a
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( 1a
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:
161 raise ResourceNotFound()
162 await customer_service.delete_payment_method(session, payment_method)