Coverage for polar/customer_portal/endpoints/subscription.py: 46%
62 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from typing import Annotated 1a
3import structlog 1a
4from fastapi import Depends, Query 1a
6from polar.exceptions import ResourceNotFound 1a
7from polar.kit.db.postgres import AsyncSession 1a
8from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
9from polar.kit.schemas import MultipleQueryFilter 1a
10from polar.kit.sorting import Sorting, SortingGetter 1a
11from polar.locker import Locker, get_locker 1a
12from polar.models import Subscription 1a
13from polar.openapi import APITag 1a
14from polar.postgres import get_db_session 1a
15from polar.product.schemas import ProductID 1a
16from polar.routing import APIRouter 1a
17from polar.subscription.schemas import SubscriptionChargePreview, SubscriptionID 1a
18from polar.subscription.service import AlreadyCanceledSubscription 1a
19from polar.subscription.service import subscription as subscription_service 1a
21from .. import auth 1a
22from ..schemas.subscription import CustomerSubscription, CustomerSubscriptionUpdate 1a
23from ..service.subscription import CustomerSubscriptionSortProperty 1a
24from ..service.subscription import ( 1a
25 customer_subscription as customer_subscription_service,
26)
28log = structlog.get_logger() 1a
30router = APIRouter(prefix="/subscriptions", tags=["subscriptions", APITag.public]) 1a
32SubscriptionNotFound = { 1a
33 "description": "Customer subscription was not found.",
34 "model": ResourceNotFound.schema(),
35}
37ListSorting = Annotated[ 1a
38 list[Sorting[CustomerSubscriptionSortProperty]],
39 Depends(SortingGetter(CustomerSubscriptionSortProperty, ["-started_at"])),
40]
43@router.get( 1a
44 "/", summary="List Subscriptions", response_model=ListResource[CustomerSubscription]
45)
46async def list( 1a
47 auth_subject: auth.CustomerPortalRead,
48 pagination: PaginationParamsQuery,
49 sorting: ListSorting,
50 product_id: MultipleQueryFilter[ProductID] | None = Query(
51 None, title="ProductID Filter", description="Filter by product ID."
52 ),
53 active: bool | None = Query(
54 None,
55 description=("Filter by active or cancelled subscription."),
56 ),
57 query: str | None = Query(
58 None, description="Search by product or organization name."
59 ),
60 session: AsyncSession = Depends(get_db_session),
61) -> ListResource[CustomerSubscription]:
62 """List subscriptions of the authenticated customer."""
63 results, count = await customer_subscription_service.list(
64 session,
65 auth_subject,
66 product_id=product_id,
67 active=active,
68 query=query,
69 pagination=pagination,
70 sorting=sorting,
71 )
73 return ListResource.from_paginated_results(
74 [CustomerSubscription.model_validate(result) for result in results],
75 count,
76 pagination,
77 )
80@router.get( 1a
81 "/{id}",
82 summary="Get Subscription",
83 response_model=CustomerSubscription,
84 responses={404: SubscriptionNotFound},
85)
86async def get( 1a
87 id: SubscriptionID,
88 auth_subject: auth.CustomerPortalRead,
89 session: AsyncSession = Depends(get_db_session),
90) -> Subscription:
91 """Get a subscription for the authenticated customer."""
92 subscription = await customer_subscription_service.get_by_id(
93 session, auth_subject, id
94 )
96 if subscription is None:
97 raise ResourceNotFound()
99 return subscription
102@router.get( 1a
103 "/{id}/charge-preview",
104 summary="Preview Next Charge For Active Subscription",
105 response_model=SubscriptionChargePreview,
106 responses={404: SubscriptionNotFound},
107 tags=[APITag.private],
108)
109async def get_charge_preview( 1a
110 id: SubscriptionID,
111 auth_subject: auth.CustomerPortalRead,
112 session: AsyncSession = Depends(get_db_session),
113) -> SubscriptionChargePreview:
114 """Get current period usage and cost breakdown for a subscription."""
115 subscription = await customer_subscription_service.get_by_id(
116 session, auth_subject, id
117 )
119 if subscription is None:
120 raise ResourceNotFound()
122 # Allow active, trialing, and subscriptions set to cancel at period end
123 if subscription.status not in ("active", "trialing"):
124 raise ResourceNotFound()
126 # If subscription will end (cancel_at_period_end or ends_at), ensure there's still a charge coming
127 if subscription.cancel_at_period_end or subscription.ends_at:
128 # Only show preview if we haven't reached the end date yet
129 if subscription.ended_at:
130 raise ResourceNotFound()
132 return await subscription_service.calculate_charge_preview(session, subscription)
135@router.patch( 1a
136 "/{id}",
137 summary="Update Subscription",
138 response_model=CustomerSubscription,
139 responses={
140 200: {"description": "Customer subscription updated."},
141 403: {
142 "description": (
143 "Customer subscription is already canceled "
144 "or will be at the end of the period."
145 ),
146 "model": AlreadyCanceledSubscription.schema(),
147 },
148 404: SubscriptionNotFound,
149 },
150)
151async def update( 1a
152 id: SubscriptionID,
153 subscription_update: CustomerSubscriptionUpdate,
154 auth_subject: auth.CustomerPortalWrite,
155 session: AsyncSession = Depends(get_db_session),
156 locker: Locker = Depends(get_locker),
157) -> Subscription:
158 """Update a subscription of the authenticated customer."""
159 subscription = await customer_subscription_service.get_by_id(
160 session, auth_subject, id
161 )
163 if subscription is None:
164 raise ResourceNotFound()
166 log.info(
167 "customer_portal.subscription.cancel",
168 id=id,
169 customer_id=auth_subject.subject.id,
170 updates=subscription_update,
171 )
172 async with subscription_service.lock(locker, subscription):
173 return await customer_subscription_service.update(
174 session, subscription, updates=subscription_update
175 )
178@router.delete( 1a
179 "/{id}",
180 summary="Cancel Subscription",
181 response_model=CustomerSubscription,
182 responses={
183 200: {"description": "Customer subscription is canceled."},
184 403: {
185 "description": (
186 "Customer subscription is already canceled "
187 "or will be at the end of the period."
188 ),
189 "model": AlreadyCanceledSubscription.schema(),
190 },
191 404: SubscriptionNotFound,
192 },
193)
194async def cancel( 1a
195 id: SubscriptionID,
196 auth_subject: auth.CustomerPortalWrite,
197 session: AsyncSession = Depends(get_db_session),
198 locker: Locker = Depends(get_locker),
199) -> Subscription:
200 """Cancel a subscription of the authenticated customer."""
201 subscription = await customer_subscription_service.get_by_id(
202 session, auth_subject, id
203 )
205 if subscription is None:
206 raise ResourceNotFound()
208 log.info(
209 "customer_portal.subscription.cancel",
210 id=id,
211 customer_id=auth_subject.subject.id,
212 )
213 async with subscription_service.lock(locker, subscription):
214 return await customer_subscription_service.cancel(session, subscription)