py(types): Add type annotations to email utils.

mypy.ini

@@ -66,6 +66,7 @@ files = src/sentry/api/bases/external_actor.py,
         src/sentry/utils/avatar.py,
         src/sentry/utils/codecs.py,
         src/sentry/utils/dates.py,
+        src/sentry/utils/email/,
         src/sentry/utils/jwt.py,
         src/sentry/utils/kvstore,
         src/sentry/utils/time_window.py,
@@ -104,6 +105,8 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 [mypy-jsonschema]
 ignore_missing_imports = True
+[mypy-lxml]
+ignore_missing_imports = True
 [mypy-mistune.*]
 ignore_missing_imports = True
 [mypy-rb.*]
@@ -112,5 +115,7 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 [mypy-sentry_relay.*]
 ignore_missing_imports = True
+[mypy-toronado]
+ignore_missing_imports = True
 [mypy-zstandard]
 ignore_missing_imports = True

src/sentry/integrations/vsts/webhooks.py

@@ -1,5 +1,4 @@
 import logging
-import re
 from typing import Any, Mapping, Optional
 
 from django.utils.crypto import constant_time_compare
@@ -11,12 +10,11 @@
 from sentry.integrations.utils import sync_group_assignee_inbound
 from sentry.models import Identity, Integration, OrganizationIntegration
 from sentry.models.apitoken import generate_token
+from sentry.utils.email import parse_email
 
 from .client import VstsApiClient
 
 UNSET = object()
-# Pull email from the string: u'lauryn <lauryn@sentry.io>'
-EMAIL_PARSER = re.compile(r"<(.*)>")
 logger = logging.getLogger("sentry.integrations")
 PROVIDER_KEY = "vsts"
 
@@ -140,7 +138,7 @@ def handle_assign_to(
         new_value = assigned_to.get("newValue")
         if new_value is not None:
             try:
-                email = self.parse_email(new_value)
+                email = parse_email(new_value)
             except AttributeError as e:
                 logger.info(
                     "vsts.failed-to-parse-email-in-handle-assign-to",
@@ -186,10 +184,6 @@ def handle_status_change(
 
             installation.sync_status_inbound(external_issue_key, data)
 
-    def parse_email(self, email: str) -> str:
-        # TODO(mgaeta): This is too brittle and doesn't pass types.
-        return EMAIL_PARSER.search(email).group(1)  # type: ignore
-
     def create_subscription(
         self, instance: Optional[str], identity_data: Mapping[str, Any], oauth_redirect_url: str
     ) -> Response:

src/sentry/utils/email/__init__.py

@@ -7,6 +7,7 @@
     "group_id_to_email",
     "inline_css",
     "is_smtp_enabled",
+    "parse_email",
     "ListResolver",
     "MessageBuilder",
     "PreviewBackend",
@@ -20,7 +21,7 @@
 user emails in the database.
 """
 
-from .address import email_to_group_id, group_id_to_email
+from .address import email_to_group_id, group_id_to_email, parse_email
 from .backend import PreviewBackend, is_smtp_enabled
 from .faker import create_fake_email
 from .list_resolver import ListResolver

src/sentry/utils/email/address.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+import re
 from email.utils import parseaddr
 
 from django.conf import settings
@@ -9,21 +12,23 @@
 
 # cache the domain_from_email calculation
 # This is just a tuple of (email, email-domain)
-_from_email_domain_cache = (None, None)
+_from_email_domain_cache: tuple[str, str] | None = None
 
+# Pull email from the string: "lauryn <lauryn@sentry.io>"
+EMAIL_PARSER = re.compile(r"<([^>]*)>")
 
 signer = _CaseInsensitiveSigner()
 
 
-def get_from_email_domain():
+def get_from_email_domain() -> str:
     global _from_email_domain_cache
     from_ = options.get("mail.from")
-    if not _from_email_domain_cache[0] == from_:
+    if _from_email_domain_cache is None or not _from_email_domain_cache[0] == from_:
         _from_email_domain_cache = (from_, domain_from_email(from_))
     return _from_email_domain_cache[1]
 
 
-def email_to_group_id(address):
+def email_to_group_id(address: str) -> int:
     """
     Email address should be in the form of:
         {group_id}+{signature}@example.com
@@ -33,7 +38,7 @@ def email_to_group_id(address):
     return int(force_bytes(signer.unsign(signed_data)))
 
 
-def group_id_to_email(group_id):
+def group_id_to_email(group_id: int) -> str:
     signed_data = signer.sign(str(group_id))
     return "@".join(
         (
@@ -43,7 +48,7 @@ def group_id_to_email(group_id):
     )
 
 
-def domain_from_email(email):
+def domain_from_email(email: str) -> str:
     email = parseaddr(email)[1]
     try:
         return email.split("@", 1)[1]
@@ -54,3 +59,8 @@ def domain_from_email(email):
 
 def is_valid_email_address(value: str) -> bool:
     return not settings.INVALID_EMAIL_ADDRESS_PATTERN.search(value)
+
+
+def parse_email(email: str) -> str:
+    matches = EMAIL_PARSER.search(email)
+    return matches.group(1) if matches else ""

src/sentry/utils/email/backend.py

@@ -1,36 +1,42 @@
+from __future__ import annotations
+
 import subprocess
 import tempfile
+from typing import Any, Sequence
 
 from django.conf import settings
+from django.core.mail import EmailMultiAlternatives
 from django.core.mail.backends.base import BaseEmailBackend
 
 from sentry import options
 
+Backend = Any
+
 
-def is_smtp_enabled(backend=None):
+def is_smtp_enabled(backend: Backend | None = None) -> bool:
     """Check if the current backend is SMTP based."""
     if backend is None:
         backend = get_mail_backend()
     return backend not in settings.SENTRY_SMTP_DISABLED_BACKENDS
 
 
-def get_mail_backend():
+def get_mail_backend() -> Backend:
     backend = options.get("mail.backend")
     try:
         return settings.SENTRY_EMAIL_BACKEND_ALIASES[backend]
     except KeyError:
         return backend
 
 
-class PreviewBackend(BaseEmailBackend):
+class PreviewBackend(BaseEmailBackend):  # type: ignore
     """
     Email backend that can be used in local development to open messages in the
     local mail client as they are sent.
 
     Probably only works on OS X.
     """
 
-    def send_messages(self, email_messages):
+    def send_messages(self, email_messages: Sequence[EmailMultiAlternatives]) -> int:
         for message in email_messages:
             content = bytes(message.message())
             preview = tempfile.NamedTemporaryFile(

src/sentry/utils/email/faker.py

@@ -3,7 +3,7 @@
 FAKE_EMAIL_TLD = ".sentry-fake"
 
 
-def create_fake_email(unique_id, namespace):
+def create_fake_email(unique_id: str, namespace: str) -> str:
     """
     Generate a fake email of the form: {unique_id}@{namespace}{FAKE_EMAIL_TLD}
 
@@ -12,6 +12,6 @@ def create_fake_email(unique_id, namespace):
     return f"{unique_id}@{namespace}{FAKE_EMAIL_TLD}"
 
 
-def is_fake_email(email):
+def is_fake_email(email: str) -> bool:
     """Returns True if the provided email matches the fake email pattern."""
     return email.endswith(FAKE_EMAIL_TLD)

src/sentry/utils/email/list_resolver.py

@@ -1,7 +1,14 @@
+from __future__ import annotations
+
+from collections import Mapping
+from typing import Callable, Generic, Iterable
+
+from sentry.db.models import Model
+from sentry.db.models.manager import M
 from sentry.utils.strings import is_valid_dot_atom
 
 
-class ListResolver:
+class ListResolver(Generic[M]):
     """
     Manages the generation of RFC 2919 compliant list-id strings from varying
     objects types.
@@ -12,7 +19,9 @@ class UnregisteredTypeError(Exception):
         Error raised when attempting to build a list-id from an unregistered object type.
         """
 
-    def __init__(self, namespace, type_handlers):
+    def __init__(
+        self, namespace: str, type_handlers: Mapping[type[Model], Callable[[M], Iterable[str]]]
+    ) -> None:
         assert is_valid_dot_atom(namespace)
 
         # The list-id-namespace that will be used when generating the list-id
@@ -26,7 +35,7 @@ def __init__(self, namespace, type_handlers):
         # values.
         self.__type_handlers = type_handlers
 
-    def __call__(self, instance):
+    def __call__(self, instance: M) -> str:
         """
         Build a list-id string from an instance.
 

src/sentry/utils/email/manager.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import Iterable, Mapping
 
@@ -8,7 +10,9 @@
 logger = logging.getLogger("sentry.mail")
 
 
-def get_email_addresses(user_ids: Iterable[int], project: Project = None) -> Mapping[int, str]:
+def get_email_addresses(
+    user_ids: Iterable[int], project: Project | None = None
+) -> Mapping[int, str]:
     """
     Find the best email addresses for a collection of users. If a project is
     provided, prefer their project-specific notification preferences.
@@ -35,7 +39,7 @@ def get_email_addresses(user_ids: Iterable[int], project: Project = None) -> Map
 
     if pending:
         logger.warning(
-            "Could not resolve email addresses for user IDs in %r, discarding...", pending
+            f"Could not resolve email addresses for user IDs in {pending}, discarding..."
         )
 
     return results