Coverage for polar/customer_portal/endpoints/subscription.py: 70%
62 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from typing import Annotated
3import structlog
4from fastapi import Depends, Query
6from polar.exceptions import ResourceNotFound
7from polar.kit.db.postgres import AsyncSession
8from polar.kit.pagination import ListResource, PaginationParamsQuery
9from polar.kit.schemas import MultipleQueryFilter
10from polar.kit.sorting import Sorting, SortingGetter
11from polar.locker import Locker, get_locker
12from polar.models import Subscription
13from polar.openapi import APITag
14from polar.postgres import get_db_session
15from polar.product.schemas import ProductID
16from polar.routing import APIRouter
17from polar.subscription.schemas import SubscriptionChargePreview, SubscriptionID
18from polar.subscription.service import AlreadyCanceledSubscription
19from polar.subscription.service import subscription as subscription_service
21from .. import auth
22from ..schemas.subscription import CustomerSubscription, CustomerSubscriptionUpdate
23from ..service.subscription import CustomerSubscriptionSortProperty
24from ..service.subscription import (
25 customer_subscription as customer_subscription_service,
26)
28log = structlog.get_logger()
30router = APIRouter(prefix="/subscriptions", tags=["subscriptions", APITag.public])
32SubscriptionNotFound = {
33 "description": "Customer subscription was not found.",
34 "model": ResourceNotFound.schema(),
35}
37ListSorting = Annotated[
38 list[Sorting[CustomerSubscriptionSortProperty]],
39 Depends(SortingGetter(CustomerSubscriptionSortProperty, ["-started_at"])),
40]
43@router.get(
44 "/", summary="List Subscriptions", response_model=ListResource[CustomerSubscription]
45)
46async def list(
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(
81 "/{id}",
82 summary="Get Subscription",
83 response_model=CustomerSubscription,
84 responses={404: SubscriptionNotFound},
85)
86async def get(
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: 96 ↛ 99line 96 didn't jump to line 99 because the condition on line 96 was always true
97 raise ResourceNotFound()
99 return subscription
102@router.get(
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(
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: 119 ↛ 123line 119 didn't jump to line 123 because the condition on line 119 was always true
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(
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(
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: 163 ↛ 166line 163 didn't jump to line 166 because the condition on line 163 was always true
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(
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(
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: 205 ↛ 208line 205 didn't jump to line 208 because the condition on line 205 was always true
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)