Coverage for polar/checkout_link/endpoints.py: 44%
71 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import Annotated 1a
3from fastapi import Depends, Path, Query, Request 1a
4from fastapi.datastructures import URL 1a
5from fastapi.responses import RedirectResponse 1a
6from pydantic import UUID4 1a
8from polar.checkout import ip_geolocation 1a
9from polar.checkout.service import checkout as checkout_service 1a
10from polar.checkout_link.repository import CheckoutLinkRepository 1a
11from polar.exceptions import ResourceNotFound 1a
12from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
13from polar.kit.schemas import MultipleQueryFilter 1a
14from polar.models import CheckoutLink 1a
15from polar.openapi import APITag 1a
16from polar.organization.schemas import OrganizationID 1a
17from polar.postgres import AsyncSession, get_db_session 1a
18from polar.product.schemas import ProductID 1a
19from polar.routing import APIRouter 1a
21from . import auth, sorting 1a
22from .schemas import CheckoutLink as CheckoutLinkSchema 1a
23from .schemas import CheckoutLinkCreate, CheckoutLinkUpdate 1a
24from .service import checkout_link as checkout_link_service 1a
26router = APIRouter(prefix="/checkout-links", tags=["checkout-links", APITag.public]) 1a
29CheckoutLinkID = Annotated[UUID4, Path(description="The checkout link ID.")] 1a
30CheckoutLinkClientSecret = Annotated[ 1a
31 str, Path(description="The checkout link client secret.")
32]
33CheckoutLinkNotFound = { 1a
34 "description": "Checkout link not found.",
35 "model": ResourceNotFound.schema(),
36}
39@router.get( 1a
40 "/", summary="List Checkout Links", response_model=ListResource[CheckoutLinkSchema]
41)
42async def list( 1a
43 auth_subject: auth.CheckoutLinkRead,
44 pagination: PaginationParamsQuery,
45 sorting: sorting.ListSorting,
46 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
47 None, title="OrganizationID Filter", description="Filter by organization ID."
48 ),
49 product_id: MultipleQueryFilter[ProductID] | None = Query(
50 None, title="ProductID Filter", description="Filter by product ID."
51 ),
52 session: AsyncSession = Depends(get_db_session),
53) -> ListResource[CheckoutLinkSchema]:
54 """List checkout links."""
55 results, count = await checkout_link_service.list(
56 session,
57 auth_subject,
58 organization_id=organization_id,
59 product_id=product_id,
60 pagination=pagination,
61 sorting=sorting,
62 )
64 return ListResource.from_paginated_results(
65 [CheckoutLinkSchema.model_validate(result) for result in results],
66 count,
67 pagination,
68 )
71@router.get( 1a
72 "/{id}",
73 summary="Get Checkout Link",
74 response_model=CheckoutLinkSchema,
75 responses={404: CheckoutLinkNotFound},
76)
77async def get( 1a
78 id: CheckoutLinkID,
79 auth_subject: auth.CheckoutLinkRead,
80 session: AsyncSession = Depends(get_db_session),
81) -> CheckoutLink:
82 """Get a checkout link by ID."""
83 checkout_link = await checkout_link_service.get_by_id(session, auth_subject, id)
85 if checkout_link is None:
86 raise ResourceNotFound()
88 return checkout_link
91@router.post( 1a
92 "/",
93 response_model=CheckoutLinkSchema,
94 status_code=201,
95 summary="Create Checkout Link",
96 responses={201: {"description": "Checkout link created."}},
97)
98async def create( 1a
99 checkout_link_create: CheckoutLinkCreate,
100 auth_subject: auth.CheckoutLinkWrite,
101 session: AsyncSession = Depends(get_db_session),
102) -> CheckoutLink:
103 """Create a checkout link."""
104 return await checkout_link_service.create(
105 session, checkout_link_create, auth_subject
106 )
109@router.patch( 1a
110 "/{id}",
111 response_model=CheckoutLinkSchema,
112 summary="Update Checkout Link",
113 responses={
114 200: {"description": "Checkout link updated."},
115 404: CheckoutLinkNotFound,
116 },
117)
118async def update( 1a
119 id: CheckoutLinkID,
120 checkout_link_update: CheckoutLinkUpdate,
121 auth_subject: auth.CheckoutLinkWrite,
122 session: AsyncSession = Depends(get_db_session),
123) -> CheckoutLink:
124 """Update a checkout link."""
125 checkout_link = await checkout_link_service.get_by_id(session, auth_subject, id)
127 if checkout_link is None:
128 raise ResourceNotFound()
130 return await checkout_link_service.update(
131 session, checkout_link, checkout_link_update, auth_subject
132 )
135@router.delete( 1a
136 "/{id}",
137 status_code=204,
138 summary="Delete Checkout Link",
139 responses={
140 204: {"description": "Checkout link deleted."},
141 404: CheckoutLinkNotFound,
142 },
143)
144async def delete( 1a
145 id: CheckoutLinkID,
146 auth_subject: auth.CheckoutLinkWrite,
147 session: AsyncSession = Depends(get_db_session),
148) -> None:
149 """Delete a checkout link."""
150 checkout_link = await checkout_link_service.get_by_id(session, auth_subject, id)
152 if checkout_link is None:
153 raise ResourceNotFound()
155 await checkout_link_service.delete(session, checkout_link)
158@router.get("/{client_secret}/redirect", include_in_schema=False) 1a
159async def redirect( 1ab
160 request: Request,
161 client_secret: CheckoutLinkClientSecret,
162 ip_geolocation_client: ip_geolocation.IPGeolocationClient,
163 embed_origin: str | None = Query(None),
164 session: AsyncSession = Depends(get_db_session),
165 # Product pre-selection & query parameter prefill
166 product_id: UUID4 | None = Query(None),
167 amount: str | None = Query(None),
168 customer_email: str | None = Query(None),
169 customer_name: str | None = Query(None),
170 discount_code: str | None = Query(None),
171 # Metadata that can be set from query parameters
172 reference_id: str | None = Query(None),
173 utm_source: str | None = Query(None),
174 utm_medium: str | None = Query(None),
175 utm_campaign: str | None = Query(None),
176 utm_term: str | None = Query(None),
177 utm_content: str | None = Query(None),
178) -> RedirectResponse:
179 """Use a checkout link to create a checkout session and redirect to it."""
180 repository = CheckoutLinkRepository.from_session(session)
181 checkout_link = await repository.get_by_client_secret(
182 client_secret, options=repository.get_eager_options()
183 )
185 if checkout_link is None:
186 raise ResourceNotFound()
188 ip_address = request.client.host if request.client else None
190 # Build query_prefill dictionary from explicit parameters
191 query_prefill: dict[str, str | UUID4 | dict[str, str] | None] = {
192 "product_id": product_id,
193 "amount": amount,
194 "customer_email": customer_email,
195 "customer_name": customer_name,
196 "discount_code": discount_code,
197 }
199 # Extract custom_field_data.* parameters from query string
200 custom_field_data: dict[str, str] = {}
201 for key, value in request.query_params.items():
202 if key.startswith("custom_field_data."):
203 slug = key.replace("custom_field_data.", "")
204 custom_field_data[slug] = value
206 if custom_field_data:
207 query_prefill["custom_field_data"] = custom_field_data
209 checkout = await checkout_service.checkout_link_create(
210 session,
211 checkout_link,
212 embed_origin,
213 ip_geolocation_client,
214 ip_address,
215 query_prefill=query_prefill,
216 reference_id=reference_id,
217 utm_source=utm_source,
218 utm_medium=utm_medium,
219 utm_campaign=utm_campaign,
220 utm_term=utm_term,
221 utm_content=utm_content,
222 )
224 validated_custom_field_keys = (
225 set(checkout.custom_field_data.keys()) if checkout.custom_field_data else set()
226 )
228 checkout_url = URL(checkout.url)
229 query_params = {
230 k: v
231 for k, v in request.query_params.items()
232 if k != "embed_origin"
233 and (k not in query_prefill or query_prefill[k] is None)
234 and not (
235 k.startswith("custom_field_data.")
236 and k.replace("custom_field_data.", "") in validated_custom_field_keys
237 )
238 }
239 checkout_url = checkout_url.include_query_params(**query_params)
241 return RedirectResponse(checkout_url)