Coverage for polar/customer_portal/endpoints/customer_seat.py: 27%
94 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
5from pydantic import UUID4 1a
6from sqlalchemy.orm import joinedload, selectinload 1a
8from polar.customer_seat.repository import CustomerSeatRepository 1a
9from polar.customer_seat.schemas import CustomerSeat as CustomerSeatSchema 1a
10from polar.customer_seat.schemas import SeatAssign, SeatsList 1a
11from polar.customer_seat.service import seat_service 1a
12from polar.exceptions import BadRequest, ResourceNotFound 1a
13from polar.kit.db.postgres import AsyncSession 1a
14from polar.models import CustomerSeat, Order, Product, Subscription 1a
15from polar.openapi import APITag 1a
16from polar.order.repository import OrderRepository 1a
17from polar.postgres import get_db_session 1a
18from polar.routing import APIRouter 1a
19from polar.subscription.repository import SubscriptionRepository 1a
21from .. import auth 1a
22from ..schemas.subscription import CustomerSubscription 1a
24log = structlog.get_logger() 1a
26router = APIRouter(prefix="/seats", tags=["seats", APITag.public]) 1a
29@router.get( 1a
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( 1a
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:
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:
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( 1a
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( 1a
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:
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:
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( 1a
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( 1a
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:
205 raise ResourceNotFound("Seat not found")
207 return await seat_service.revoke_seat(session, seat)
210@router.post( 1a
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( 1a
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:
230 raise ResourceNotFound("Seat not found")
232 return await seat_service.resend_invitation(session, seat)
235@router.get( 1a
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( 1a
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)