Skip to content

Commit

Permalink
Improve type hints in hc.lib
Browse files Browse the repository at this point in the history
  • Loading branch information
cuu508 committed Sep 8, 2023
1 parent 9173475 commit 92bf977
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 27 deletions.
43 changes: 23 additions & 20 deletions hc/lib/emails.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,17 @@
from email.utils import make_msgid
from smtplib import SMTPDataError, SMTPServerDisconnected
from threading import Thread
from typing import Any

from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.core.mail import EmailMultiAlternatives as Message
from django.template.loader import render_to_string as render


class EmailThread(Thread):
MAX_TRIES = 3

def __init__(self, message) -> None:
def __init__(self, message: Message) -> None:
Thread.__init__(self)
self.message = message

Expand All @@ -35,7 +36,9 @@ def run(self) -> None:
time.sleep(1)


def make_message(name, to, ctx, headers={}):
def make_message(
name: str, to: str | list[str], ctx: dict[str, Any], headers: dict[str, str] = {}
) -> Message:
subject = render("emails/%s-subject.html" % name, ctx).strip()
body = render("emails/%s-body-text.html" % name, ctx)
html = render("emails/%s-body-html.html" % name, ctx)
Expand All @@ -55,18 +58,18 @@ def make_message(name, to, ctx, headers={}):
from_email = settings.DEFAULT_FROM_EMAIL

to_list = [to] if isinstance(to, str) else to
msg = EmailMultiAlternatives(subject, body, from_email, to_list, headers=headers)
msg = Message(subject, body, from_email, to_list, headers=headers)
msg.attach_alternative(html, "text/html")
return msg


def send(msg, block: bool = False) -> None:
def send(message: Message, block: bool = False) -> None:
assert settings.EMAIL_HOST, (
"No SMTP configuration,"
" see https://github.com/healthchecks/healthchecks#sending-emails"
)

t = EmailThread(msg)
t = EmailThread(message)
if block or hasattr(settings, "BLOCKING_EMAILS"):
# In tests, we send emails synchronously
# so we can inspect the outgoing messages
Expand All @@ -77,53 +80,53 @@ def send(msg, block: bool = False) -> None:
t.start()


def login(to, ctx) -> None:
def login(to: str, ctx: dict[str, Any]) -> None:
send(make_message("login", to, ctx))


def transfer_request(to, ctx) -> None:
def transfer_request(to: str, ctx: dict[str, Any]) -> None:
send(make_message("transfer-request", to, ctx))


def alert(to, ctx, headers={}) -> None:
def alert(to: str, ctx: dict[str, Any], headers: dict[str, str]) -> None:
send(make_message("alert", to, ctx, headers=headers))


def verify_email(to, ctx) -> None:
def verify_email(to: str, ctx: dict[str, Any]) -> None:
send(make_message("verify-email", to, ctx))


def report(to, ctx, headers={}) -> None:
def report(to: str, ctx: dict[str, Any], headers: dict[str, str]) -> None:
m = make_message("report", to, ctx, headers=headers)
send(m, block=True)


def nag(to, ctx, headers={}) -> None:
def nag(to: str, ctx: dict[str, Any], headers: dict[str, str]) -> None:
m = make_message("nag", to, ctx, headers=headers)
send(m, block=True)


def deletion_notice(to, ctx, headers={}) -> None:
m = make_message("deletion-notice", to, ctx, headers=headers)
def deletion_notice(to: str, ctx: dict[str, Any]) -> None:
m = make_message("deletion-notice", to, ctx)
send(m, block=True)


def deletion_scheduled(to, ctx, headers={}) -> None:
m = make_message("deletion-scheduled", to, ctx, headers=headers)
def deletion_scheduled(to: list[str], ctx: dict[str, Any]) -> None:
m = make_message("deletion-scheduled", to, ctx)
send(m, block=True)


def sms_limit(to, ctx) -> None:
def sms_limit(to: str, ctx: dict[str, Any]) -> None:
send(make_message("sms-limit", to, ctx))


def call_limit(to, ctx) -> None:
def call_limit(to: str, ctx: dict[str, Any]) -> None:
send(make_message("phone-call-limit", to, ctx))


def sudo_code(to, ctx) -> None:
def sudo_code(to: str, ctx: dict[str, Any]) -> None:
send(make_message("sudo-code", to, ctx))


def signal_rate_limited(to, ctx) -> None:
def signal_rate_limited(to: str, ctx: dict[str, Any]) -> None:
send(make_message("signal-rate-limited", to, ctx))
15 changes: 8 additions & 7 deletions hc/lib/html.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,28 +3,29 @@
import re
from html import unescape
from html.parser import HTMLParser
from typing import Any


class TextOnlyParser(HTMLParser):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __init__(self) -> None:
super().__init__()
self.active = True
self.buf = []
self.buf: list[str] = []
self.skiplist = set(["script", "style"])

def handle_starttag(self, tag, attrs):
def handle_starttag(self, tag: str, attrs: Any) -> None:
if tag in self.skiplist:
self.active = False

def handle_endtag(self, tag):
def handle_endtag(self, tag: str) -> None:
if tag in self.skiplist:
self.active = True

def handle_data(self, data):
def handle_data(self, data: str) -> None:
if self.active and data:
self.buf.append(data)

def get_text(self):
def get_text(self) -> str:
messy = "".join(self.buf)
return " ".join(messy.split())

Expand Down

0 comments on commit 92bf977

Please sign in to comment.