Coverage for polar/customer_portal/endpoints/customer_session.py: 96%

1from fastapi import Depends 

2 

3from polar.kit.db.postgres import AsyncSession 

4from polar.models import CustomerSession 

5from polar.openapi import APITag 

6from polar.postgres import get_db_session 

7from polar.routing import APIRouter 

8 

9from .. import auth 

10from ..schemas.customer_session import ( 

11 CustomerCustomerSession, 

12 CustomerSessionCodeAuthenticateRequest, 

13 CustomerSessionCodeAuthenticateResponse, 

14 CustomerSessionCodeInvalidOrExpiredResponse, 

15 CustomerSessionCodeRequest, 

16) 

17from ..service.customer_session import ( 

18 CustomerDoesNotExist, 

19 OrganizationDoesNotExist, 

20) 

21from ..service.customer_session import customer_session as customer_session_service 

22 

23router = APIRouter(prefix="/customer-session", tags=["customer-session"]) 

24 

25 

26@router.post( 

27 "/request", 

28 name="customer_portal.customer_session.request", 

29 status_code=202, 

30 tags=[APITag.private], 

31) 

32async def request( 

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 

45 

46 await customer_session_service.send( 

47 session, 

48 customer_session_code, 

49 code, 

50 ) 

51 

52 

53@router.post( 

54 "/authenticate", 

55 name="customer_portal.customer_session.authenticate", 

56 responses={ 

57 401: CustomerSessionCodeInvalidOrExpiredResponse, 

58 }, 

59 tags=[APITag.private], 

60) 

61async def authenticate( 

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) 

69 

70 

71@router.get( 

72 "/introspect", 

73 summary="Introspect Customer Session", 

74 tags=[APITag.public], 

75 response_model=CustomerCustomerSession, 

76) 

77async def introspect( 

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