Coverage for polar/login_code/endpoints.py: 64%

1from urllib.parse import urlencode 

2 

3from fastapi import Depends, Form, Request 

4from fastapi.responses import RedirectResponse 

5 

6from polar.auth.dependencies import WebUserOrAnonymous 

7from polar.auth.models import is_user 

8from polar.auth.service import auth as auth_service 

9from polar.config import settings 

10from polar.integrations.loops.service import loops as loops_service 

11from polar.kit.db.postgres import AsyncSession 

12from polar.kit.email import EmailStrDNS 

13from polar.kit.http import ReturnTo 

14from polar.openapi import APITag 

15from polar.postgres import get_db_session 

16from polar.posthog import posthog 

17from polar.routing import APIRouter 

18 

19from .schemas import LoginCodeRequest 

20from .service import LoginCodeError 

21from .service import login_code as login_code_service 

22 

23router = APIRouter(prefix="/login-code", tags=["login_code", APITag.private]) 

24 

25 

26@router.post("/request", status_code=202) 

27async def request_login_code( 

28 login_code_request: LoginCodeRequest, 

29 session: AsyncSession = Depends(get_db_session), 

30) -> None: 

31 """ 

32 Request a login code. 

33 """ 

34 code_model, code = await login_code_service.request( 

35 session, 

36 login_code_request.email, 

37 return_to=login_code_request.return_to, 

38 signup_attribution=login_code_request.attribution, 

39 ) 

40 

41 # Send the code email 

42 await login_code_service.send(code_model, code) 

43 

44 

45@router.post("/authenticate") 

46async def authenticate_login_code( 

47 request: Request, 

48 return_to: ReturnTo, 

49 email: EmailStrDNS, 

50 auth_subject: WebUserOrAnonymous, 

51 code: str = Form(), 

52 session: AsyncSession = Depends(get_db_session), 

53) -> RedirectResponse: 

54 """ 

55 Authenticate with a login code. 

56 """ 

57 if is_user(auth_subject): 57 ↛ 60line 57 didn't jump to line 60 because the condition on line 57 was always true

58 return RedirectResponse(return_to, 303) 

59 

60 try: 

61 user, is_signup = await login_code_service.authenticate( 

62 session, code=code, email=email 

63 ) 

64 except LoginCodeError as e: 

65 base_url = str(settings.generate_frontend_url("/login/code/verify")) 

66 url_params = { 

67 "return_to": return_to, 

68 "email": request.query_params.get("email"), 

69 "error": e.message, 

70 } 

71 failed_login_return_to = f"{base_url}?{urlencode(url_params)}" 

72 return RedirectResponse(failed_login_return_to, 303) 

73 

74 # Event tracking last to ensure business critical data is stored first 

75 if is_signup: 

76 posthog.user_signup(user, "code") 

77 await loops_service.user_signup(user, emailLogin=True) 

78 else: 

79 posthog.user_login(user, "code") 

80 await loops_service.user_update(session, user, emailLogin=True) 

81 

82 return await auth_service.get_login_response( 

83 session, request, user, return_to=return_to 

84 )