Coverage for polar/login_code/endpoints.py: 52%
40 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from urllib.parse import urlencode 1a
3from fastapi import Depends, Form, Request 1a
4from fastapi.responses import RedirectResponse 1a
6from polar.auth.dependencies import WebUserOrAnonymous 1a
7from polar.auth.models import is_user 1a
8from polar.auth.service import auth as auth_service 1a
9from polar.config import settings 1a
10from polar.integrations.loops.service import loops as loops_service 1a
11from polar.kit.db.postgres import AsyncSession 1a
12from polar.kit.email import EmailStrDNS 1a
13from polar.kit.http import ReturnTo 1a
14from polar.openapi import APITag 1a
15from polar.postgres import get_db_session 1a
16from polar.posthog import posthog 1a
17from polar.routing import APIRouter 1a
19from .schemas import LoginCodeRequest 1a
20from .service import LoginCodeError 1a
21from .service import login_code as login_code_service 1a
23router = APIRouter(prefix="/login-code", tags=["login_code", APITag.private]) 1a
26@router.post("/request", status_code=202) 1a
27async def request_login_code( 1ab
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 )
41 # Send the code email
42 await login_code_service.send(code_model, code)
45@router.post("/authenticate") 1a
46async def authenticate_login_code( 1a
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):
58 return RedirectResponse(return_to, 303)
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)
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)
82 return await auth_service.get_login_response(
83 session, request, user, return_to=return_to
84 )