zerver/lib: Use python 3 syntax for typing.

Edited by tabbott to improve various line-wrapping decisions.
This commit is contained in:
rht
2017-11-05 11:15:10 +01:00
committed by Tim Abbott
parent 229a8b38c0
commit ee546a33a3
8 changed files with 523 additions and 746 deletions

View File

@@ -32,8 +32,7 @@ import ujson
import urllib
from collections import defaultdict
def one_click_unsubscribe_link(user_profile, email_type):
# type: (UserProfile, str) -> str
def one_click_unsubscribe_link(user_profile: UserProfile, email_type: str) -> str:
"""
Generate a unique link that a logged-out user can visit to unsubscribe from
Zulip e-mails without having to first log in.
@@ -42,33 +41,28 @@ def one_click_unsubscribe_link(user_profile, email_type):
Confirmation.UNSUBSCRIBE,
url_args = {'email_type': email_type})
def hash_util_encode(string):
# type: (Text) -> Text
def hash_util_encode(string: Text) -> Text:
# Do the same encoding operation as hash_util.encodeHashComponent on the
# frontend.
# `safe` has a default value of "/", but we want those encoded, too.
return urllib.parse.quote(
string.encode("utf-8"), safe=b"").replace(".", "%2E").replace("%", ".")
def pm_narrow_url(realm, participants):
# type: (Realm, List[Text]) -> Text
def pm_narrow_url(realm: Realm, participants: List[Text]) -> Text:
participants.sort()
base_url = u"%s/#narrow/pm-with/" % (realm.uri,)
return base_url + hash_util_encode(",".join(participants))
def stream_narrow_url(realm, stream):
# type: (Realm, Text) -> Text
def stream_narrow_url(realm: Realm, stream: Text) -> Text:
base_url = u"%s/#narrow/stream/" % (realm.uri,)
return base_url + hash_util_encode(stream)
def topic_narrow_url(realm, stream, topic):
# type: (Realm, Text, Text) -> Text
def topic_narrow_url(realm: Realm, stream: Text, topic: Text) -> Text:
base_url = u"%s/#narrow/stream/" % (realm.uri,)
return u"%s%s/topic/%s" % (base_url, hash_util_encode(stream),
hash_util_encode(topic))
def relative_to_full_url(base_url, content):
# type: (Text, Text) -> Text
def relative_to_full_url(base_url: Text, content: Text) -> Text:
# Convert relative URLs to absolute URLs.
fragment = lxml.html.fromstring(content)
@@ -101,10 +95,8 @@ def relative_to_full_url(base_url, content):
return content
def fix_emojis(content, base_url, emojiset):
# type: (Text, Text, Text) -> Text
def make_emoji_img_elem(emoji_span_elem):
# type: (Any) -> Dict[str, Any]
def fix_emojis(content: Text, base_url: Text, emojiset: Text) -> Text:
def make_emoji_img_elem(emoji_span_elem: Any) -> Dict[str, Any]:
# Convert the emoji spans to img tags.
classes = emoji_span_elem.get('class')
match = re.search('emoji-(?P<emoji_code>\S+)', classes)
@@ -138,8 +130,7 @@ def fix_emojis(content, base_url, emojiset):
content = lxml.html.tostring(fragment).decode('utf-8')
return content
def build_message_list(user_profile, messages):
# type: (UserProfile, List[Message]) -> List[Dict[str, Any]]
def build_message_list(user_profile: UserProfile, messages: List[Message]) -> List[Dict[str, Any]]:
"""
Builds the message list object for the missed message email template.
The messages are collapsed into per-recipient and per-sender blocks, like
@@ -147,22 +138,19 @@ def build_message_list(user_profile, messages):
"""
messages_to_render = [] # type: List[Dict[str, Any]]
def sender_string(message):
# type: (Message) -> Text
def sender_string(message: Message) -> Text:
if message.recipient.type in (Recipient.STREAM, Recipient.HUDDLE):
return message.sender.full_name
else:
return ''
def fix_plaintext_image_urls(content):
# type: (Text) -> Text
def fix_plaintext_image_urls(content: Text) -> Text:
# Replace image URLs in plaintext content of the form
# [image name](image url)
# with a simple hyperlink.
return re.sub(r"\[(\S*)\]\((\S*)\)", r"\2", content)
def build_message_payload(message):
# type: (Message) -> Dict[str, Text]
def build_message_payload(message: Message) -> Dict[str, Text]:
plain = message.content
plain = fix_plaintext_image_urls(plain)
# There's a small chance of colliding with non-Zulip URLs containing
@@ -181,14 +169,12 @@ def build_message_list(user_profile, messages):
return {'plain': plain, 'html': html}
def build_sender_payload(message):
# type: (Message) -> Dict[str, Any]
def build_sender_payload(message: Message) -> Dict[str, Any]:
sender = sender_string(message)
return {'sender': sender,
'content': [build_message_payload(message)]}
def message_header(user_profile, message):
# type: (UserProfile, Message) -> Dict[str, Any]
def message_header(user_profile: UserProfile, message: Message) -> Dict[str, Any]:
disp_recipient = get_display_recipient(message.recipient)
if message.recipient.type == Recipient.PERSONAL:
header = u"You and %s" % (message.sender.full_name,)
@@ -264,8 +250,9 @@ def build_message_list(user_profile, messages):
return messages_to_render
@statsd_increment("missed_message_reminders")
def do_send_missedmessage_events_reply_in_zulip(user_profile, missed_messages, message_count):
# type: (UserProfile, List[Message], int) -> None
def do_send_missedmessage_events_reply_in_zulip(user_profile: UserProfile,
missed_messages: List[Message],
message_count: int) -> None:
"""
Send a reminder email to a user if she's missed some PMs by being offline.
@@ -384,8 +371,7 @@ def do_send_missedmessage_events_reply_in_zulip(user_profile, missed_messages, m
user_profile.last_reminder = timezone_now()
user_profile.save(update_fields=['last_reminder'])
def handle_missedmessage_emails(user_profile_id, missed_email_events):
# type: (int, Iterable[Dict[str, Any]]) -> None
def handle_missedmessage_emails(user_profile_id: int, missed_email_events: Iterable[Dict[str, Any]]) -> None:
message_ids = [event.get('message_id') for event in missed_email_events]
user_profile = get_user_profile_by_id(user_profile_id)
@@ -429,29 +415,25 @@ def handle_missedmessage_emails(user_profile_id, missed_email_events):
message_count_by_recipient_subject[recipient_subject],
)
def clear_scheduled_invitation_emails(email):
# type: (str) -> None
def clear_scheduled_invitation_emails(email: str) -> None:
"""Unlike most scheduled emails, invitation emails don't have an
existing user object to key off of, so we filter by address here."""
items = ScheduledEmail.objects.filter(address__iexact=email,
type=ScheduledEmail.INVITATION_REMINDER)
items.delete()
def clear_scheduled_emails(user_id, email_type=None):
# type: (int, Optional[int]) -> None
def clear_scheduled_emails(user_id: int, email_type: Optional[int]=None) -> None:
items = ScheduledEmail.objects.filter(user_id=user_id)
if email_type is not None:
items = items.filter(type=email_type)
items.delete()
def log_digest_event(msg):
# type: (Text) -> None
def log_digest_event(msg: Text) -> None:
import logging
logging.basicConfig(filename=settings.DIGEST_LOG_PATH, level=logging.INFO)
logging.info(msg)
def enqueue_welcome_emails(user):
# type: (UserProfile) -> None
def enqueue_welcome_emails(user: UserProfile) -> None:
from zerver.context_processors import common_context
if settings.WELCOME_EMAIL_SENDER is not None:
# line break to avoid triggering lint rule
@@ -476,8 +458,7 @@ def enqueue_welcome_emails(user):
"zerver/emails/followup_day2", to_user_id=user.id, from_name=from_name,
from_address=from_address, context=context, delay=datetime.timedelta(days=1))
def convert_html_to_markdown(html):
# type: (Text) -> Text
def convert_html_to_markdown(html: Text) -> Text:
# On Linux, the tool installs as html2markdown, and there's a command called
# html2text that does something totally different. On OSX, the tool installs
# as html2text.