Coverage for polar/customer_portal/endpoints/customer_seat.py: 66%
94 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
5from pydantic import UUID4
6from sqlalchemy.orm import joinedload, selectinload
8from polar.customer_seat.repository import CustomerSeatRepository
9from polar.customer_seat.schemas import CustomerSeat as CustomerSeatSchema
10from polar.customer_seat.schemas import SeatAssign, SeatsList
11from polar.customer_seat.service import seat_service
12from polar.exceptions import BadRequest, ResourceNotFound
13from polar.kit.db.postgres import AsyncSession
14from polar.models import CustomerSeat, Order, Product, Subscription
15from polar.openapi import APITag
16from polar.order.repository import OrderRepository
17from polar.postgres import get_db_session
18from polar.routing import APIRouter
19from polar.subscription.repository import SubscriptionRepository
21from .. import auth
22from ..schemas.subscription import CustomerSubscription
24log = structlog.get_logger()
26router = APIRouter(prefix="/seats", tags=["seats", APITag.public])
29@router.get(
30 "",
31 summary="List Seats",
32 response_model=SeatsList,
33 responses={
34 401: {"description": "Authentication required"},
35 403: {"description": "Not permitted or seat-based pricing not enabled"},
36 404: {"description": "Subscription or order not found"},
37 },
38)
39async def list_seats(
40 auth_subject: auth.CustomerPortalRead,
41 session: AsyncSession = Depends(get_db_session),
42 subscription_id: Annotated[
43 UUID4 | None, Query(description="Subscription ID")
44 ] = None,
45 order_id: Annotated[UUID4 | None, Query(description="Order ID")] = None,
46) -> SeatsList:
47 customer = auth_subject.subject
49 subscription: Subscription | None = None
50 order: Order | None = None
51 total_seats = 0
53 if subscription_id:
54 subscription_repository = SubscriptionRepository.from_session(session)
56 statement = (
57 subscription_repository.get_readable_statement(auth_subject)
58 .options(*subscription_repository.get_eager_options())
59 .where(
60 Subscription.id == subscription_id,
61 )
62 )
63 subscription = await subscription_repository.get_one_or_none(statement)
65 if not subscription: 65 ↛ 68line 65 didn't jump to line 68 because the condition on line 65 was always true
66 raise ResourceNotFound("Subscription not found")
68 total_seats = subscription.seats or 0
70 elif order_id:
71 order_repository = OrderRepository.from_session(session)
73 order_statement = (
74 order_repository.get_readable_statement(auth_subject)
75 .options(*order_repository.get_eager_options())
76 .where(
77 Order.id == order_id,
78 )
79 )
80 order = await order_repository.get_one_or_none(order_statement)
82 if not order: 82 ↛ 85line 82 didn't jump to line 85 because the condition on line 82 was always true
83 raise ResourceNotFound("Order not found")
85 total_seats = order.seats or 0
87 else:
88 raise BadRequest("Either subscription_id or order_id must be provided")
90 container = subscription or order
91 assert container is not None # Already validated above
93 seats = await seat_service.list_seats(session, container)
94 available_seats = await seat_service.get_available_seats_count(session, container)
96 return SeatsList(
97 seats=[CustomerSeatSchema.model_validate(seat) for seat in seats],
98 available_seats=available_seats,
99 total_seats=total_seats,
100 )
103@router.post(
104 "",
105 summary="Assign Seat",
106 response_model=CustomerSeatSchema,
107 responses={
108 400: {"description": "No available seats or customer already has a seat"},
109 401: {"description": "Authentication required"},
110 403: {"description": "Not permitted or seat-based pricing not enabled"},
111 404: {"description": "Subscription, order, or customer not found"},
112 },
113)
114async def assign_seat(
115 seat_assign: SeatAssign,
116 auth_subject: auth.CustomerPortalWrite,
117 session: AsyncSession = Depends(get_db_session),
118) -> CustomerSeat:
119 customer = auth_subject.subject
121 subscription: Subscription | None = None
122 order: Order | None = None
124 if seat_assign.subscription_id: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 subscription_repository = SubscriptionRepository.from_session(session)
127 statement = (
128 subscription_repository.get_base_statement()
129 .options(*subscription_repository.get_eager_options())
130 .where(
131 Subscription.id == seat_assign.subscription_id,
132 Subscription.customer_id == customer.id,
133 )
134 )
135 subscription = await subscription_repository.get_one_or_none(statement)
137 if not subscription:
138 raise ResourceNotFound("Subscription not found")
140 elif seat_assign.order_id: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 order_repository = OrderRepository.from_session(session)
143 order_statement = (
144 order_repository.get_base_statement()
145 .options(*order_repository.get_eager_options())
146 .where(
147 Order.id == seat_assign.order_id,
148 Order.customer_id == customer.id,
149 )
150 )
151 order = await order_repository.get_one_or_none(order_statement)
153 if not order:
154 raise ResourceNotFound("Order not found")
156 else:
157 raise BadRequest("Either subscription_id or order_id is required")
159 container = subscription or order
160 assert container is not None # Already validated above
162 seat = await seat_service.assign_seat(
163 session,
164 container,
165 email=seat_assign.email,
166 external_customer_id=seat_assign.external_customer_id,
167 customer_id=seat_assign.customer_id,
168 metadata=seat_assign.metadata,
169 )
171 # Reload seat with customer relationship
172 seat_repository = CustomerSeatRepository.from_session(session)
173 seat_statement = (
174 seat_repository.get_base_statement()
175 .where(CustomerSeat.id == seat.id)
176 .options(joinedload(CustomerSeat.customer))
177 )
178 reloaded_seat = await seat_repository.get_one_or_none(seat_statement)
180 if not reloaded_seat:
181 raise ResourceNotFound("Seat not found after creation")
183 return reloaded_seat
186@router.delete(
187 "/{seat_id}",
188 summary="Revoke Seat",
189 response_model=CustomerSeatSchema,
190 responses={
191 401: {"description": "Authentication required"},
192 403: {"description": "Not permitted or seat-based pricing not enabled"},
193 404: {"description": "Seat not found"},
194 },
195)
196async def revoke_seat(
197 seat_id: UUID4,
198 auth_subject: auth.CustomerPortalWrite,
199 session: AsyncSession = Depends(get_db_session),
200) -> CustomerSeat:
201 customer = auth_subject.subject
203 seat = await seat_service.get_seat_for_customer(session, customer, seat_id)
204 if not seat: 204 ↛ 207line 204 didn't jump to line 207 because the condition on line 204 was always true
205 raise ResourceNotFound("Seat not found")
207 return await seat_service.revoke_seat(session, seat)
210@router.post(
211 "/{seat_id}/resend",
212 summary="Resend Invitation",
213 response_model=CustomerSeatSchema,
214 responses={
215 400: {"description": "Seat is not pending or already claimed"},
216 401: {"description": "Authentication required"},
217 403: {"description": "Not permitted or seat-based pricing not enabled"},
218 404: {"description": "Seat not found"},
219 },
220)
221async def resend_invitation(
222 seat_id: UUID4,
223 auth_subject: auth.CustomerPortalWrite,
224 session: AsyncSession = Depends(get_db_session),
225) -> CustomerSeat:
226 customer = auth_subject.subject
228 seat = await seat_service.get_seat_for_customer(session, customer, seat_id)
229 if not seat: 229 ↛ 232line 229 didn't jump to line 232 because the condition on line 229 was always true
230 raise ResourceNotFound("Seat not found")
232 return await seat_service.resend_invitation(session, seat)
235@router.get(
236 "/subscriptions",
237 summary="List Claimed Subscriptions",
238 response_model=list[CustomerSubscription],
239 responses={
240 401: {"description": "Authentication required"},
241 },
242)
243async def list_claimed_subscriptions(
244 auth_subject: auth.CustomerPortalRead,
245 session: AsyncSession = Depends(get_db_session),
246) -> list[Subscription]:
247 """List all subscriptions where the authenticated customer has claimed a seat."""
248 subscription_repository = SubscriptionRepository.from_session(session)
250 statement = subscription_repository.get_claimed_subscriptions_statement(
251 auth_subject
252 ).options(
253 joinedload(Subscription.customer),
254 joinedload(Subscription.product).options(
255 selectinload(Product.product_medias),
256 joinedload(Product.organization),
257 ),
258 )
260 subscriptions = await subscription_repository.get_all(statement)
262 return list(subscriptions)