Coverage for polar/integrations/apple/endpoints.py: 32%
73 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1import uuid 1a
2from typing import Any 1a
4from fastapi import Depends, Form, Request 1a
5from fastapi.responses import RedirectResponse 1a
6from httpx_oauth.integrations.fastapi import ( 1a
7 OAuth2AuthorizeCallbackError,
8)
9from httpx_oauth.oauth2 import GetAccessTokenError 1a
11from polar.auth.dependencies import WebUserOrAnonymous 1a
12from polar.auth.models import is_user 1a
13from polar.auth.service import auth as auth_service 1a
14from polar.config import settings 1a
15from polar.exceptions import PolarRedirectionError 1a
16from polar.integrations.loops.service import loops as loops_service 1a
17from polar.kit import jwt 1a
18from polar.kit.http import ReturnTo 1a
19from polar.openapi import APITag 1a
20from polar.postgres import AsyncSession, get_db_session 1a
21from polar.posthog import posthog 1a
22from polar.routing import APIRouter 1a
23from polar.user.schemas import UserSignupAttribution, UserSignupAttributionQuery 1a
25from .service import AppleServiceError, get_apple_oauth_client 1a
26from .service import apple as apple_service 1a
29class OAuthCallbackError(PolarRedirectionError): ... 1a
32router = APIRouter( 1a
33 prefix="/integrations/apple",
34 tags=["integrations_apple", APITag.private],
35)
38@router.get("/authorize", name="integrations.apple.authorize") 1a
39async def apple_authorize( 1a
40 request: Request,
41 auth_subject: WebUserOrAnonymous,
42 return_to: ReturnTo,
43 signup_attribution: UserSignupAttributionQuery,
44) -> RedirectResponse:
45 state: dict[str, Any] = {}
47 state["return_to"] = return_to
49 if signup_attribution:
50 state["signup_attribution"] = signup_attribution.model_dump(exclude_unset=True)
52 if is_user(auth_subject):
53 state["user_id"] = str(auth_subject.subject.id)
55 encoded_state = jwt.encode(data=state, secret=settings.SECRET, type="apple_oauth")
56 redirect_uri = str(request.url_for("integrations.apple.callback"))
57 apple_oauth_client = get_apple_oauth_client()
58 authorization_url = await apple_oauth_client.get_authorization_url(
59 redirect_uri=redirect_uri,
60 state=encoded_state,
61 extras_params={"response_mode": "form_post"},
62 )
63 return RedirectResponse(authorization_url, 303)
66@router.post("/callback", name="integrations.apple.callback") 1a
67async def apple_callback( 1a
68 request: Request,
69 auth_subject: WebUserOrAnonymous,
70 code: str | None = Form(None),
71 code_verifier: str | None = Form(None),
72 state: str | None = Form(None),
73 error: str | None = Form(None),
74 session: AsyncSession = Depends(get_db_session),
75) -> RedirectResponse:
76 if code is None or error is not None:
77 raise OAuth2AuthorizeCallbackError(
78 status_code=400,
79 detail=error if error is not None else None,
80 )
82 redirect_uri = str(request.url_for("integrations.apple.callback"))
83 try:
84 apple_oauth_client = get_apple_oauth_client(secret=True)
85 token_data = await apple_oauth_client.get_access_token(
86 code, redirect_uri, code_verifier
87 )
88 except GetAccessTokenError as e:
89 raise OAuth2AuthorizeCallbackError(
90 status_code=500,
91 detail=e.message,
92 response=e.response,
93 ) from e
95 error_description = token_data.get("error_description")
96 if error_description:
97 raise OAuthCallbackError(error_description)
98 if not state:
99 raise OAuthCallbackError("No state")
101 try:
102 state_data = jwt.decode(token=state, secret=settings.SECRET, type="apple_oauth")
103 except jwt.DecodeError as e:
104 raise OAuthCallbackError("Invalid state") from e
106 return_to = state_data.get("return_to", None)
107 state_user_id = state_data.get("user_id")
109 state_signup_attribution = state_data.get("signup_attribution")
110 if state_signup_attribution:
111 state_signup_attribution = UserSignupAttribution.model_validate(
112 state_signup_attribution
113 )
115 try:
116 if (
117 is_user(auth_subject)
118 and state_user_id is not None
119 and auth_subject.subject.id == uuid.UUID(state_user_id)
120 ):
121 is_signup = False
122 user = await apple_service.link_user(
123 session, user=auth_subject.subject, token=token_data
124 )
125 else:
126 user, is_signup = await apple_service.get_updated_or_create(
127 session,
128 token=token_data,
129 signup_attribution=state_signup_attribution,
130 )
131 except AppleServiceError as e:
132 raise OAuthCallbackError(e.message, e.status_code, return_to=return_to) from e
134 # Event tracking last to ensure business critical data is stored first
135 if is_signup:
136 posthog.user_signup(user, "apple")
137 await loops_service.user_signup(user)
138 else:
139 posthog.user_login(user, "apple")
140 await loops_service.user_update(session, user)
142 return await auth_service.get_login_response(
143 session, request, user, return_to=return_to
144 )