Coverage for polar/customer_portal/endpoints/customer_session.py: 63%
27 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from fastapi import Depends 1a
3from polar.kit.db.postgres import AsyncSession 1a
4from polar.models import CustomerSession 1a
5from polar.openapi import APITag 1a
6from polar.postgres import get_db_session 1a
7from polar.routing import APIRouter 1a
9from .. import auth 1a
10from ..schemas.customer_session import ( 1a
11 CustomerCustomerSession,
12 CustomerSessionCodeAuthenticateRequest,
13 CustomerSessionCodeAuthenticateResponse,
14 CustomerSessionCodeInvalidOrExpiredResponse,
15 CustomerSessionCodeRequest,
16)
17from ..service.customer_session import ( 1a
18 CustomerDoesNotExist,
19 OrganizationDoesNotExist,
20)
21from ..service.customer_session import customer_session as customer_session_service 1a
23router = APIRouter(prefix="/customer-session", tags=["customer-session"]) 1a
26@router.post( 1a
27 "/request",
28 name="customer_portal.customer_session.request",
29 status_code=202,
30 tags=[APITag.private],
31)
32async def request( 1ab
33 customer_session_code_request: CustomerSessionCodeRequest,
34 session: AsyncSession = Depends(get_db_session),
35) -> None:
36 try:
37 customer_session_code, code = await customer_session_service.request(
38 session,
39 customer_session_code_request.email,
40 customer_session_code_request.organization_id,
41 )
42 except (CustomerDoesNotExist, OrganizationDoesNotExist):
43 # We don't want to leak information about whether the customer or organization exists
44 return
46 await customer_session_service.send(
47 session,
48 customer_session_code,
49 code,
50 )
53@router.post( 1a
54 "/authenticate",
55 name="customer_portal.customer_session.authenticate",
56 responses={
57 401: CustomerSessionCodeInvalidOrExpiredResponse,
58 },
59 tags=[APITag.private],
60)
61async def authenticate( 1ab
62 authenticated_request: CustomerSessionCodeAuthenticateRequest,
63 session: AsyncSession = Depends(get_db_session),
64) -> CustomerSessionCodeAuthenticateResponse:
65 token, _ = await customer_session_service.authenticate(
66 session, authenticated_request.code
67 )
68 return CustomerSessionCodeAuthenticateResponse(token=token)
71@router.get( 1a
72 "/introspect",
73 summary="Introspect Customer Session",
74 tags=[APITag.public],
75 response_model=CustomerCustomerSession,
76)
77async def introspect( 1a
78 auth_subject: auth.CustomerPortalRead,
79) -> CustomerSession:
80 """Introspect the current session and return its information."""
81 session = auth_subject.session
82 assert isinstance(session, CustomerSession)
83 return session