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

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.api.base import Endpoint
 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 @@ class WorkItemWebhook(Endpoint):  # type: ignore
         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 @@ class WorkItemWebhook(Endpoint):  # type: ignore
 
             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 @@ __all__ = (
     "group_id_to_email",
     "inline_css",
     "is_smtp_enabled",
+    "parse_email",
     "ListResolver",
     "MessageBuilder",
     "PreviewBackend",
@@ -20,7 +21,7 @@ parsing email address strings, building and sending messages, and looking up
 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 @@ from .signer import _CaseInsensitiveSigner
 
 # 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,20 +1,26 @@
+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]
@@ -22,7 +28,7 @@ def get_mail_backend():
         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.
@@ -30,7 +36,7 @@ class PreviewBackend(BaseEmailBackend):
     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 ListResolver:
         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 @@ class ListResolver:
         # 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 @@ from .faker import is_fake_email
 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

src/sentry/utils/email/message_builder.py

@@ -1,10 +1,12 @@
+from __future__ import annotations
+
 import logging
 import os
 import time
 from functools import partial
 from operator import attrgetter
 from random import randrange
-from typing import Iterable, Optional
+from typing import Any, Callable, Iterable, Mapping, MutableMapping, Sequence
 
 import lxml
 import toronado
@@ -12,6 +14,7 @@ from django.core.mail import EmailMultiAlternatives
 from django.utils.encoding import force_text
 
 from sentry import options
+from sentry.db.models import Model
 from sentry.logging import LoggingFormat
 from sentry.models import Activity, Group, GroupEmailThread, Project
 from sentry.utils import metrics
@@ -25,7 +28,7 @@ from .send import send_messages
 
 logger = logging.getLogger("sentry.mail")
 
-default_list_type_handlers = {
+default_list_type_handlers: Mapping[type[Model], Callable[[Model], Iterable[str]]] = {
     Activity: attrgetter("project.slug", "project.organization.slug"),
     Project: attrgetter("slug", "organization.slug"),
     Group: attrgetter("project.slug", "organization.slug"),
@@ -42,7 +45,7 @@ MAX_RECIPIENTS = 5
 # Slightly modified version of Django's `django.core.mail.message:make_msgid`
 # because we need to override the domain. If we ever upgrade to django 1.8, we
 # can/should replace this.
-def make_msgid(domain):
+def make_msgid(domain: str) -> str:
     """Returns a string suitable for RFC 2822 compliant Message-ID, e.g:
     <20020201195627.33539.96671@nightshade.la.mastaler.com>
     Optional idstring if given is a string used to strengthen the
@@ -63,75 +66,78 @@ def inline_css(value: str) -> str:
     toronado.inline(tree)
     # CSS media query support is inconsistent when the DOCTYPE declaration is
     # missing, so we force it to HTML5 here.
-    return lxml.html.tostring(tree, doctype="<!DOCTYPE html>", encoding=None).decode("utf-8")
+    html: str = lxml.html.tostring(tree, doctype="<!DOCTYPE html>", encoding=None).decode("utf-8")
+    return html
 
 
 class MessageBuilder:
     def __init__(
         self,
-        subject,
-        context=None,
-        template=None,
-        html_template=None,
-        body="",
-        html_body=None,
-        headers=None,
-        reference=None,
-        reply_reference=None,
-        from_email=None,
-        type=None,
-    ):
+        subject: str,
+        context: Mapping[str, Any] | None = None,
+        template: str | None = None,
+        html_template: str | None = None,
+        body: str = "",
+        html_body: str | None = None,
+        headers: Mapping[str, str] | None = None,
+        reference: Model | None = None,
+        reply_reference: Model | None = None,
+        from_email: str | None = None,
+        type: str | None = None,
+    ) -> None:
         assert not (body and template)
         assert not (html_body and html_template)
         assert context or not (template or html_template)
 
-        if headers is None:
-            headers = {}
-
         self.subject = subject
         self.context = context or {}
         self.template = template
         self.html_template = html_template
         self._txt_body = body
         self._html_body = html_body
-        self.headers = headers
+        self.headers: MutableMapping[str, Any] = {**(headers or {})}
         self.reference = reference  # The object that generated this message
         self.reply_reference = reply_reference  # The object this message is replying about
         self.from_email = from_email or options.get("mail.from")
-        self._send_to = set()
+        self._send_to: set[str] = set()
         self.type = type if type else "generic"
 
-        if reference is not None and "List-Id" not in headers:
+        if reference is not None and "List-Id" not in self.headers:
             try:
-                headers["List-Id"] = make_listid_from_instance(reference)
+                self.headers["List-Id"] = make_listid_from_instance(reference)
             except ListResolver.UnregisteredTypeError as error:
                 logger.debug(str(error))
             except AssertionError as error:
                 logger.warning(str(error))
 
-    def __render_html_body(self) -> str:
-        html_body = None
+    def __render_html_body(self) -> str | None:
         if self.html_template:
             html_body = render_to_string(self.html_template, self.context)
         else:
             html_body = self._html_body
 
-        if html_body is not None:
-            return inline_css(html_body)
+        if html_body is None:
+            return None
+
+        return inline_css(html_body)
 
     def __render_text_body(self) -> str:
         if self.template:
-            return render_to_string(self.template, self.context)
+            body: str = render_to_string(self.template, self.context)
+            return body
         return self._txt_body
 
-    def add_users(self, user_ids: Iterable[int], project: Optional[Project] = None) -> None:
+    def add_users(self, user_ids: Iterable[int], project: Project | None = None) -> None:
         self._send_to.update(list(get_email_addresses(user_ids, project).values()))
 
-    def build(self, to, reply_to=None, cc=None, bcc=None):
-        if self.headers is None:
-            headers = {}
-        else:
-            headers = self.headers.copy()
+    def build(
+        self,
+        to: str,
+        reply_to: Iterable[str] | None = None,
+        cc: Iterable[str] | None = None,
+        bcc: Iterable[str] | None = None,
+    ) -> EmailMultiAlternatives:
+        headers = {**self.headers}
 
         if options.get("mail.enable-replies") and "X-Sentry-Reply-To" in headers:
             reply_to = headers["X-Sentry-Reply-To"]
@@ -151,7 +157,7 @@ class MessageBuilder:
 
         if self.reply_reference is not None:
             reference = self.reply_reference
-            subject = "Re: %s" % subject
+            subject = f"Re: {subject}"
         else:
             reference = self.reference
 
@@ -181,7 +187,12 @@ class MessageBuilder:
 
         return msg
 
-    def get_built_messages(self, to=None, cc=None, bcc=None):
+    def get_built_messages(
+        self,
+        to: Iterable[str] | None = None,
+        cc: Iterable[str] | None = None,
+        bcc: Iterable[str] | None = None,
+    ) -> Sequence[EmailMultiAlternatives]:
         send_to = set(to or ())
         send_to.update(self._send_to)
         results = [
@@ -191,27 +202,38 @@ class MessageBuilder:
             logger.debug("Did not build any messages, no users to send to.")
         return results
 
-    def format_to(self, to):
+    def format_to(self, to: list[str]) -> str:
         if not to:
             return ""
         if len(to) > MAX_RECIPIENTS:
             to = to[:MAX_RECIPIENTS] + [f"and {len(to[MAX_RECIPIENTS:])} more."]
         return ", ".join(to)
 
-    def send(self, to=None, cc=None, bcc=None, fail_silently=False):
+    def send(
+        self,
+        to: Iterable[str] | None = None,
+        cc: Iterable[str] | None = None,
+        bcc: Iterable[str] | None = None,
+        fail_silently: bool = False,
+    ) -> int:
         return send_messages(
             self.get_built_messages(to, cc=cc, bcc=bcc), fail_silently=fail_silently
         )
 
-    def send_async(self, to=None, cc=None, bcc=None):
+    def send_async(
+        self,
+        to: Iterable[str] | None = None,
+        cc: Iterable[str] | None = None,
+        bcc: Iterable[str] | None = None,
+    ) -> None:
         from sentry.tasks.email import send_email
 
         fmt = options.get("system.logging-format")
         messages = self.get_built_messages(to, cc=cc, bcc=bcc)
-        extra = {"message_type": self.type}
+        extra: MutableMapping[str, str | tuple[str]] = {"message_type": self.type}
         loggable = [v for k, v in self.context.items() if hasattr(v, "id")]
         for context in loggable:
-            extra["%s_id" % type(context).__name__.lower()] = context.id
+            extra[f"{type(context).__name__.lower()}_id"] = context.id
 
         log_mail_queued = partial(logger.info, "mail.queued", extra=extra)
         for message in messages:

src/sentry/utils/email/send.py

@@ -1,6 +1,8 @@
 import logging
+from typing import Any, Sequence
 
 from django.core import mail
+from django.core.mail import EmailMultiAlternatives
 
 from sentry import options
 from sentry.utils import metrics
@@ -10,9 +12,10 @@ from .backend import get_mail_backend
 logger = logging.getLogger("sentry.mail")
 
 
-def send_messages(messages, fail_silently=False):
+def send_messages(messages: Sequence[EmailMultiAlternatives], fail_silently: bool = False) -> int:
     connection = get_connection(fail_silently=fail_silently)
-    sent = connection.send_messages(messages)
+    # Explicitly typing to satisfy mypy.
+    sent: int = connection.send_messages(messages)
     metrics.incr("email.sent", len(messages), skip_internal=False)
     for message in messages:
         extra = {
@@ -23,7 +26,7 @@ def send_messages(messages, fail_silently=False):
     return sent
 
 
-def get_connection(fail_silently=False):
+def get_connection(fail_silently: bool = False) -> Any:
     """Gets an SMTP connection using our OptionsStore."""
     return mail.get_connection(
         backend=get_mail_backend(),
@@ -38,17 +41,25 @@ def get_connection(fail_silently=False):
     )
 
 
-def send_mail(subject, message, from_email, recipient_list, fail_silently=False, **kwargs):
+def send_mail(
+    subject: str,
+    message: str,
+    from_email: str,
+    recipient_list: Sequence[str],
+    fail_silently: bool = False,
+    **kwargs: Any,
+) -> int:
     """
     Wrapper that forces sending mail through our connection.
     Uses EmailMessage class which has more options than the simple send_mail
     """
-    email = mail.EmailMessage(
+    # Explicitly typing to satisfy mypy.
+    sent: int = mail.EmailMessage(
         subject,
         message,
         from_email,
         recipient_list,
         connection=get_connection(fail_silently=fail_silently),
         **kwargs,
-    )
-    return email.send(fail_silently=fail_silently)
+    ).send(fail_silently=fail_silently)
+    return sent