Coverage for polar/integrations/github/service/user.py: 29%
95 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from typing import TYPE_CHECKING, TypeAlias 1a
3import structlog 1a
4from httpx_oauth.oauth2 import OAuth2Token 1a
6from polar.exceptions import PolarError 1a
7from polar.integrations.github.client import GitHub, TokenAuthStrategy 1a
8from polar.models import OAuthAccount, User 1a
9from polar.models.user import OAuthPlatform 1a
10from polar.postgres import AsyncSession 1a
11from polar.user.oauth_service import oauth_account_service 1a
12from polar.user.repository import UserRepository 1a
13from polar.user.schemas import UserSignupAttribution 1a
14from polar.worker import enqueue_job 1a
16if TYPE_CHECKING: 16 ↛ 17line 16 didn't jump to line 17 because the condition on line 16 was never true1a
17 from githubkit.versions.latest.models import PrivateUser, PublicUser
19from .. import client as github 1a
21log = structlog.get_logger() 1a
24GithubUser: TypeAlias = "PrivateUser | PublicUser" 1a
25GithubEmail: TypeAlias = tuple[str, bool] 1a
28class GithubUserServiceError(PolarError): ... 1a
31class NoPrimaryEmailError(GithubUserServiceError): 1a
32 def __init__(self) -> None: 1a
33 super().__init__("GitHub user without primary email set")
36class CannotLinkUnverifiedEmailError(GithubUserServiceError): 1a
37 def __init__(self, email: str) -> None: 1a
38 message = (
39 f"An account already exists on Polar under the email {email}. "
40 "We cannot automatically link it to your GitHub account since "
41 "this email address is not verified on GitHub. "
42 "Either verify your email address on GitHub and try again "
43 "or sign in using your email."
44 )
45 super().__init__(message, 403)
48class AccountLinkedToAnotherUserError(GithubUserServiceError): 1a
49 def __init__(self) -> None: 1a
50 message = (
51 "This GitHub account is already linked to another user on Polar. "
52 "You may have already created another account "
53 "with a different email address."
54 )
55 super().__init__(message, 403)
58class GithubUserService: 1a
59 async def get_updated_or_create( 1a
60 self,
61 session: AsyncSession,
62 *,
63 token: OAuth2Token,
64 signup_attribution: UserSignupAttribution | None = None,
65 ) -> tuple[User, bool]:
66 client = github.get_client(access_token=token["access_token"])
67 authenticated = await self._fetch_authenticated_user(client=client)
68 user_repository = UserRepository.from_session(session)
69 user = await user_repository.get_by_oauth_account(
70 OAuthPlatform.github, str(authenticated.id)
71 )
73 if user is not None:
74 oauth_account = user.get_oauth_account(OAuthPlatform.github)
75 assert oauth_account is not None
76 oauth_account.access_token = token["access_token"]
77 oauth_account.expires_at = token["expires_at"]
78 oauth_account.account_username = authenticated.login
79 session.add(oauth_account)
80 return (user, False)
82 email, email_verified = await self.fetch_authenticated_user_primary_email(
83 client=client
84 )
86 oauth_account = OAuthAccount(
87 platform=OAuthPlatform.github,
88 account_id=str(authenticated.id),
89 account_email=email,
90 account_username=authenticated.login,
91 access_token=token["access_token"],
92 expires_at=token["expires_at"],
93 )
95 user = await user_repository.get_by_email(email)
96 if user is not None:
97 if email_verified:
98 user.oauth_accounts.append(oauth_account)
99 session.add(user)
100 return (user, False)
101 else:
102 raise CannotLinkUnverifiedEmailError(email)
104 user = User(
105 email=email,
106 email_verified=email_verified,
107 avatar_url=authenticated.avatar_url,
108 oauth_accounts=[oauth_account],
109 signup_attribution=signup_attribution,
110 )
112 session.add(user)
113 await session.flush()
115 enqueue_job("user.on_after_signup", user_id=user.id)
117 return (user, True)
119 async def link_user( 1a
120 self, session: AsyncSession, *, user: User, token: OAuth2Token
121 ) -> User:
122 client = github.get_client(access_token=token["access_token"])
123 github_user = await self._fetch_authenticated_user(client=client)
124 email, _ = await self.fetch_authenticated_user_primary_email(client=client)
126 account_id = str(github_user.id)
128 oauth_account = await oauth_account_service.get_by_platform_and_account_id(
129 session, OAuthPlatform.github, account_id
130 )
131 if oauth_account is not None:
132 if oauth_account.user_id != user.id:
133 raise AccountLinkedToAnotherUserError()
134 else:
135 oauth_account = OAuthAccount(
136 platform=OAuthPlatform.github,
137 account_id=account_id,
138 account_email=email,
139 user_id=user.id,
140 )
141 session.add(oauth_account)
142 log.info(
143 "oauth_account.connect",
144 user_id=user.id,
145 platform="github",
146 account_email=email,
147 )
149 oauth_account.access_token = token["access_token"]
150 oauth_account.expires_at = token["expires_at"]
151 oauth_account.account_email = email
152 oauth_account.account_username = github_user.login
153 session.add(oauth_account)
155 user.avatar_url = github_user.avatar_url
156 session.add(user)
158 return user
160 async def fetch_authenticated_user_primary_email( 1a
161 self, *, client: GitHub[TokenAuthStrategy]
162 ) -> GithubEmail:
163 email_response = (
164 await client.rest.users.async_list_emails_for_authenticated_user()
165 )
167 try:
168 github.ensure_expected_response(email_response)
169 except Exception as e:
170 log.error("fetch_authenticated_user_primary_email.failed", err=e)
171 raise NoPrimaryEmailError() from e
173 emails = email_response.parsed_data
175 for email in emails:
176 if email.primary:
177 return email.email, email.verified
179 raise NoPrimaryEmailError()
181 async def _fetch_authenticated_user( 1a
182 self, *, client: GitHub[TokenAuthStrategy]
183 ) -> GithubUser:
184 response = await client.rest.users.async_get_authenticated()
185 github.ensure_expected_response(response)
186 return response.parsed_data
189github_user = GithubUserService() 1a