Coverage for polar/discount/endpoints.py: 57%
43 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from typing import Annotated 1a
3from fastapi import Body, Depends, Path, Query 1a
4from pydantic import UUID4 1a
6from polar.exceptions import ResourceNotFound 1a
7from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
8from polar.kit.schemas import MultipleQueryFilter, SetSchemaReference 1a
9from polar.models import Discount 1a
10from polar.openapi import APITag 1a
11from polar.organization.schemas import OrganizationID 1a
12from polar.postgres import AsyncSession, get_db_session 1a
13from polar.routing import APIRouter 1a
15from . import auth, sorting 1a
16from .schemas import Discount as DiscountSchema 1a
17from .schemas import ( 1a
18 DiscountAdapter,
19 DiscountCreate,
20 DiscountUpdate,
21)
22from .service import discount as discount_service 1a
24router = APIRouter(prefix="/discounts", tags=["discounts", APITag.public]) 1a
27DiscountID = Annotated[UUID4, Path(description="The discount ID.")] 1a
28DiscountNotFound = { 1a
29 "description": "Discount not found.",
30 "model": ResourceNotFound.schema(),
31}
34@router.get("/", summary="List Discounts", response_model=ListResource[DiscountSchema]) 1a
35async def list( 1a
36 auth_subject: auth.DiscountRead,
37 pagination: PaginationParamsQuery,
38 sorting: sorting.ListSorting,
39 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
40 None, title="OrganizationID Filter", description="Filter by organization ID."
41 ),
42 query: str | None = Query(None, description="Filter by name."),
43 session: AsyncSession = Depends(get_db_session),
44) -> ListResource[DiscountSchema]:
45 """List discounts."""
46 results, count = await discount_service.list(
47 session,
48 auth_subject,
49 organization_id=organization_id,
50 query=query,
51 pagination=pagination,
52 sorting=sorting,
53 )
55 return ListResource.from_paginated_results(
56 [DiscountAdapter.validate_python(result) for result in results],
57 count,
58 pagination,
59 )
62@router.get( 1a
63 "/{id}",
64 summary="Get Discount",
65 response_model=DiscountSchema,
66 responses={404: DiscountNotFound},
67)
68async def get( 1a
69 id: DiscountID,
70 auth_subject: auth.DiscountRead,
71 session: AsyncSession = Depends(get_db_session),
72) -> Discount:
73 """Get a discount by ID."""
74 discount = await discount_service.get_by_id(session, auth_subject, id)
76 if discount is None:
77 raise ResourceNotFound()
79 return discount
82@router.post( 1a
83 "/",
84 response_model=DiscountSchema,
85 status_code=201,
86 summary="Create Discount",
87 responses={201: {"description": "Discount created."}},
88)
89async def create( 1a
90 auth_subject: auth.DiscountWrite,
91 # This is a workaround for FastAPI bug: https://github.com/fastapi/fastapi/discussions/12941
92 discount_create: Annotated[
93 DiscountCreate,
94 Body(...),
95 SetSchemaReference("DiscountCreate"),
96 ],
97 session: AsyncSession = Depends(get_db_session),
98) -> Discount:
99 """Create a discount."""
100 return await discount_service.create(session, discount_create, auth_subject)
103@router.patch( 1a
104 "/{id}",
105 response_model=DiscountSchema,
106 summary="Update Discount",
107 responses={
108 200: {"description": "Discount updated."},
109 404: DiscountNotFound,
110 },
111)
112async def update( 1a
113 id: DiscountID,
114 discount_update: DiscountUpdate,
115 auth_subject: auth.DiscountWrite,
116 session: AsyncSession = Depends(get_db_session),
117) -> Discount:
118 """Update a discount."""
119 discount = await discount_service.get_by_id(session, auth_subject, id)
121 if discount is None:
122 raise ResourceNotFound()
124 return await discount_service.update(session, discount, discount_update)
127@router.delete( 1a
128 "/{id}",
129 status_code=204,
130 summary="Delete Discount",
131 responses={
132 204: {"description": "Discount deleted."},
133 404: DiscountNotFound,
134 },
135)
136async def delete( 1a
137 id: DiscountID,
138 auth_subject: auth.DiscountWrite,
139 session: AsyncSession = Depends(get_db_session),
140) -> None:
141 """Delete a discount."""
142 discount = await discount_service.get_by_id(session, auth_subject, id)
144 if discount is None:
145 raise ResourceNotFound()
147 await discount_service.delete(session, discount)