|
@@ -1,21 +1,21 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
import abc
|
|
|
from dataclasses import dataclass
|
|
|
-from typing import Iterable, Optional, Sequence, Tuple, Type
|
|
|
+from typing import Collection, Iterable, Type
|
|
|
|
|
|
from sentry.models import Organization, OrganizationMember, User, UserEmail
|
|
|
from sentry.utils import metrics
|
|
|
|
|
|
|
|
|
class AmbiguousUserFromEmail(Exception):
|
|
|
- def __init__(self, email: str, users: Sequence[User]) -> None:
|
|
|
+ def __init__(self, email: str, users: Collection[User]) -> None:
|
|
|
super().__init__(f"Resolved {email!r} to {[user.id for user in users]}")
|
|
|
- self.email: str = email
|
|
|
- self.users: Tuple[User] = tuple(users)
|
|
|
+ self.email = email
|
|
|
+ self.users = tuple(users)
|
|
|
|
|
|
|
|
|
-def resolve_email_to_user(
|
|
|
- email: str, organization: Optional[Organization] = None
|
|
|
-) -> Optional[User]:
|
|
|
+def resolve_email_to_user(email: str, organization: Organization | None = None) -> User | None:
|
|
|
candidates = tuple(UserEmail.objects.filter(email__iexact=email, user__is_active=True))
|
|
|
if not candidates:
|
|
|
return None
|
|
@@ -25,9 +25,9 @@ def resolve_email_to_user(
|
|
|
@dataclass
|
|
|
class _EmailResolver:
|
|
|
email: str
|
|
|
- organization: Optional[Organization]
|
|
|
+ organization: Organization | None
|
|
|
|
|
|
- def resolve(self, candidates: Sequence[UserEmail]) -> User:
|
|
|
+ def resolve(self, candidates: Collection[UserEmail]) -> User:
|
|
|
"""Pick the user best matching the email address."""
|
|
|
|
|
|
if not candidates:
|
|
@@ -52,35 +52,35 @@ class _EmailResolver:
|
|
|
self.if_inconclusive(candidates)
|
|
|
raise AmbiguousUserFromEmail(self.email, [ue.user for ue in candidates])
|
|
|
|
|
|
- def if_inconclusive(self, remaining_candidates: Sequence[UserEmail]) -> None:
|
|
|
+ def if_inconclusive(self, remaining_candidates: Collection[UserEmail]) -> None:
|
|
|
"""Hook to call if no step resolves to a single user."""
|
|
|
metrics.incr("auth.email_resolution.no_resolution", sample_rate=1.0)
|
|
|
|
|
|
@dataclass
|
|
|
class ResolutionStep(abc.ABC):
|
|
|
- parent: "_EmailResolver"
|
|
|
+ parent: _EmailResolver
|
|
|
|
|
|
@abc.abstractmethod
|
|
|
- def apply(self, candidates: Sequence[UserEmail]) -> Iterable[UserEmail]:
|
|
|
+ def apply(self, candidates: Collection[UserEmail]) -> Iterable[UserEmail]:
|
|
|
raise NotImplementedError
|
|
|
|
|
|
- def if_conclusive(self, candidates: Sequence[UserEmail], choice: UserEmail) -> None:
|
|
|
+ def if_conclusive(self, candidates: Collection[UserEmail], choice: UserEmail) -> None:
|
|
|
"""Hook to call if this step resolves to a single user."""
|
|
|
pass
|
|
|
|
|
|
class IsVerified(ResolutionStep):
|
|
|
"""Prefer verified email addresses."""
|
|
|
|
|
|
- def apply(self, candidates: Sequence[UserEmail]) -> Iterable[UserEmail]:
|
|
|
+ def apply(self, candidates: Collection[UserEmail]) -> Iterable[UserEmail]:
|
|
|
return (ue for ue in candidates if ue.is_verified)
|
|
|
|
|
|
- def if_conclusive(self, candidates: Sequence[UserEmail], choice: UserEmail) -> None:
|
|
|
+ def if_conclusive(self, candidates: Collection[UserEmail], choice: UserEmail) -> None:
|
|
|
metrics.incr("auth.email_resolution.by_verification", sample_rate=1.0)
|
|
|
|
|
|
class HasOrgMembership(ResolutionStep):
|
|
|
"""Prefer users who belong to the organization."""
|
|
|
|
|
|
- def apply(self, candidates: Sequence[UserEmail]) -> Iterable[UserEmail]:
|
|
|
+ def apply(self, candidates: Collection[UserEmail]) -> Iterable[UserEmail]:
|
|
|
if not self.parent.organization:
|
|
|
return ()
|
|
|
query = OrganizationMember.objects.filter(
|
|
@@ -89,19 +89,19 @@ class _EmailResolver:
|
|
|
users_in_org = set(query.values_list("user", flat=True))
|
|
|
return (ue for ue in candidates if ue.user.id in users_in_org)
|
|
|
|
|
|
- def if_conclusive(self, candidates: Sequence[UserEmail], choice: UserEmail) -> None:
|
|
|
+ def if_conclusive(self, candidates: Collection[UserEmail], choice: UserEmail) -> None:
|
|
|
metrics.incr("auth.email_resolution.by_org_membership", sample_rate=1.0)
|
|
|
|
|
|
class IsPrimary(ResolutionStep):
|
|
|
"""Prefer users whose primary address matches the address in question."""
|
|
|
|
|
|
- def apply(self, candidates: Sequence[UserEmail]) -> Iterable[UserEmail]:
|
|
|
+ def apply(self, candidates: Collection[UserEmail]) -> Iterable[UserEmail]:
|
|
|
return (ue for ue in candidates if ue.is_primary())
|
|
|
|
|
|
- def if_conclusive(self, candidates: Sequence[UserEmail], choice: UserEmail) -> None:
|
|
|
+ def if_conclusive(self, candidates: Collection[UserEmail], choice: UserEmail) -> None:
|
|
|
metrics.incr("auth.email_resolution.by_primary_email", sample_rate=1.0)
|
|
|
|
|
|
- def get_steps(self) -> Iterable[Type]:
|
|
|
+ def get_steps(self) -> Iterable[Type[ResolutionStep]]:
|
|
|
return (
|
|
|
self.IsVerified,
|
|
|
self.HasOrgMembership,
|