chore: Add pytype

.gitignore

@@ -59,3 +59,4 @@ generated/
 docker-compose.override.yml
 celerybeat-schedule.*
 .coverage
+.pytype

app/api/attendees.py

@@ -285,7 +285,7 @@ def before_update_object(self, obj, data, kwargs):
                 checkout_times = (
                     obj.checkout_times.split(',') if obj.checkout_times else []
                 )
-                checkout_times.append(str(datetime.utcnow()))
+                checkout_times.append(str(datetime.datetime.utcnow()))
                 data['checkout_times'] = ','.join(checkout_times)
 
         if 'attendee_notes' in data:

app/api/helpers/import_helpers.py

@@ -369,6 +369,7 @@ def import_event_json(task_handle, zip_path, creator_id):
     except Exception as e:
         raise make_error('event', er=e)
     # create other services
+    item = []  # TODO: Remove workaround for pytype
     try:
         service_ids = {}
         for item in IMPORT_SERIES:
@@ -384,7 +385,7 @@ def import_event_json(task_handle, zip_path, creator_id):
     except IOError:
         db.session.delete(new_event)
         db.session.commit()
-        raise NotFoundError('File %s missing in event zip' % item[0])
+        raise NotFoundError('file', 'File %s missing in event zip' % item[0])
     except ValueError:
         db.session.delete(new_event)
         db.session.commit()

app/api/helpers/jwt.py

@@ -1,4 +1,4 @@
-from flask import _app_ctx_stack as ctx_stack
+from flask import _app_ctx_stack as ctx_stack  # pytype: disable=import-error
 from flask_jwt_extended.config import config
 from flask_jwt_extended.exceptions import JWTExtendedException, UserLoadError
 from flask_jwt_extended.view_decorators import _decode_jwt_from_request, _load_user

app/api/helpers/permission_manager.py

@@ -1,3 +1,4 @@
+from typing import Union
 from flask import request
 from flask_jwt_extended import current_user, verify_jwt_in_request
 from sqlalchemy.orm.exc import NoResultFound
@@ -357,7 +358,7 @@ def create_event(view, view_args, view_kwargs, *args, **kwargs):
 }
 
 
-def is_multiple(data):
+def is_multiple(data: Union[str, list]) -> bool:
     if type(data) is list:
         return True
     if type(data) is str:
@@ -465,8 +466,10 @@ def permission_manager(view, view_args, view_kwargs, *args, **kwargs):
             if not is_multiple(model):
                 model = [model]
 
-            if is_multiple(fetch_key_url):
-                fetch_key_url = fetch_key_url.split(",")
+            if type(fetch_key_url) == str and is_multiple(fetch_key_url):
+                fetch_key_url = fetch_key_url.split(  # pytype: disable=attribute-error
+                    ","
+                )
 
             found = False
             for index, mod in enumerate(model):
@@ -477,7 +480,7 @@ def permission_manager(view, view_args, view_kwargs, *args, **kwargs):
                 if not view_kwargs.get(f_url):
                     continue
                 try:
-                    data = mod.query.filter(
+                    data = mod.query.filter(  # pytype: disable=attribute-error
                         getattr(mod, fetch_key_model) == view_kwargs[f_url]
                     ).one()
                 except NoResultFound:

app/api/helpers/tasks.py

@@ -80,31 +80,31 @@ def empty_attachments_send(mail_client, message):
 
 @celery.task(name='send.email.post.sendgrid')
 def send_email_task_sendgrid(payload, headers, smtp_config):
-    try:
-        message = Mail(
-            from_email=From(payload['from'], payload['fromname']),
-            to_emails=payload['to'],
-            subject=payload['subject'],
-            html_content=payload["html"],
-        )
-        if payload['attachments'] is not None:
-            for attachment in payload['attachments']:
-                with open(attachment, 'rb') as f:
-                    file_data = f.read()
-                    f.close()
-                encoded = base64.b64encode(file_data).decode()
-                attachment = Attachment()
-                attachment.file_content = FileContent(encoded)
-                attachment.file_type = FileType('application/pdf')
-                attachment.file_name = FileName(payload['to'])
-                attachment.disposition = Disposition('attachment')
-                message.add_attachment(attachment)
-        sendgrid_client = SendGridAPIClient(get_settings()['sendgrid_key'])
-        logging.info(
-            'Sending an email regarding {} on behalf of {}'.format(
-                payload["subject"], payload["from"]
-            )
+    message = Mail(
+        from_email=From(payload['from'], payload['fromname']),
+        to_emails=payload['to'],
+        subject=payload['subject'],
+        html_content=payload["html"],
+    )
+    if payload['attachments'] is not None:
+        for attachment in payload['attachments']:
+            with open(attachment, 'rb') as f:
+                file_data = f.read()
+                f.close()
+            encoded = base64.b64encode(file_data).decode()
+            attachment = Attachment()
+            attachment.file_content = FileContent(encoded)
+            attachment.file_type = FileType('application/pdf')
+            attachment.file_name = FileName(payload['to'])
+            attachment.disposition = Disposition('attachment')
+            message.add_attachment(attachment)
+    sendgrid_client = SendGridAPIClient(get_settings()['sendgrid_key'])
+    logging.info(
+        'Sending an email regarding {} on behalf of {}'.format(
+            payload["subject"], payload["from"]
         )
+    )
+    try:
         sendgrid_client.send(message)
         logging.info('Email sent successfully')
     except urllib.error.HTTPError as e:
@@ -134,16 +134,16 @@ def send_email_task_smtp(payload, smtp_config, headers=None):
         }
     }
 
+    mailer = Mailer(mailer_config)
+    mailer.start()
+    message = Message(author=payload['from'], to=payload['to'])
+    message.subject = payload['subject']
+    message.plain = strip_tags(payload['html'])
+    message.rich = payload['html']
+    if payload['attachments'] is not None:
+        for attachment in payload['attachments']:
+            message.attach(name=attachment)
     try:
-        mailer = Mailer(mailer_config)
-        mailer.start()
-        message = Message(author=payload['from'], to=payload['to'])
-        message.subject = payload['subject']
-        message.plain = strip_tags(payload['html'])
-        message.rich = payload['html']
-        if payload['attachments'] is not None:
-            for attachment in payload['attachments']:
-                message.attach(name=attachment)
         mailer.send(message)
         logging.info('Message sent via SMTP')
     except urllib.error.HTTPError as e:

app/api/helpers/third_party_auth.py

@@ -124,7 +124,7 @@ def get_access_token(self, oauth_verifier, oauth_token):
 
     def get_authorized_client(self, oauth_verifier, oauth_token):
         resp, content = self.get_access_token(oauth_verifier, oauth_token)
-        access_token = dict(urllib.parse.urlparse.parse_qsl(content))
+        access_token = dict(urllib.parse.parse_qsl(content))
         token = oauth2.Token(
             access_token["oauth_token"], access_token["oauth_token_secret"]
         )

app/api/helpers/ticketing.py

@@ -33,7 +33,9 @@ def match_discount_quantity(discount_code, tickets=None, ticket_holders=None):
         qty = 0
         ticket_ids = [ticket.id for ticket in discount_code.tickets]
         old_holders = get_count(
-            TicketHolder.query.filter(TicketHolder.ticket_id.in_(ticket_ids))
+            TicketHolder.query.filter(
+                TicketHolder.ticket_id.in_(ticket_ids)  # pytype: disable=attribute-error
+            )
             .join(Order)
             .filter(Order.status.in_(['completed', 'placed']))
         )

app/api/helpers/utilities.py

@@ -84,9 +84,9 @@ def is_downloadable(url):
     header = h.headers
     content_type = header.get('content-type')
     # content_length = header.get('content-length', 1e10)
-    if 'text' in content_type.lower():
+    if content_type and 'text' in content_type.lower():
         return False
-    if 'html' in content_type.lower():
+    if content_type and 'html' in content_type.lower():
         return False
     return True
 

app/api/schema/events.py

@@ -34,7 +34,7 @@ def validate_timezone(self, data, original_data):
                 data['timezone'] = event.timezone
         try:
             timezone(data['timezone'])
-        except pytz.exceptions.UnknownTimeZoneError:
+        except pytz.UnknownTimeZoneError:
             raise UnprocessableEntity(
                 {'pointer': '/data/attributes/timezone'},
                 "Unknown timezone: '{}'".format(data['timezone']),

app/instance.py

@@ -182,7 +182,9 @@ def create_app():
 
         add_engine_pidguard(db.engine)
 
-        if app.config['SQLALCHEMY_DATABASE_URI'].startswith("sqlite://"):
+        if app.config[
+            'SQLALCHEMY_DATABASE_URI'  # pytype: disable=attribute-error
+        ].startswith("sqlite://"):
             sqlite_datetime_fix()
 
     sa.orm.configure_mappers()

app/models/event.py

@@ -465,7 +465,7 @@ def calc_tickets_sold_count(self):
 
     def calc_tickets_sold_prev_month(self):
         """Calculate tickets sold in the previous month"""
-        previous_month = datetime.datetime.now().month - 1
+        previous_month = datetime.now().month - 1
         return self.tickets_sold_object.filter_by(completed_at=previous_month).count()
 
     def calc_total_tickets_count(self):

app/models/helpers/versioning.py

@@ -7,11 +7,11 @@
 from bleach.callbacks import nofollow, target_blank
 
 
-def remove_line_breaks(target_string):
+def remove_line_breaks(target_string: str) -> str:
     return target_string.replace('\r', '')
 
 
-def strip_line_breaks(target_string):
+def strip_line_breaks(target_string: str) -> str:
     return target_string.replace('\n', '').replace('\r', '')
 
 
@@ -46,116 +46,14 @@ def clean_html(html):
     attrs = {'*': ['style']}
     styles = ['text-align', 'font-weight', 'text-decoration']
     cleaned = bleach.clean(html, tags=tags, attributes=attrs, styles=styles, strip=True)
-    return bleach.linkify(cleaned, callbacks=[nofollow, target_blank], parse_email=True)
+    return bleach.linkify(
+        cleaned,
+        callbacks=[nofollow, target_blank],
+        parse_email=True,  # pytype: disable=wrong-arg-types
+    )
 
 
 def strip_tags(html):
     if html is None:
         return None
     return bleach.clean(html, tags=[], attributes={}, styles=[], strip=True)
-
-
-def side_by_side_diff(old_text, new_text):
-    """
-    Calculates a side-by-side line-based difference view.
-
-    Wraps insertions in <ins></ins> and deletions in <del></del>.
-
-    From: http://code.activestate.com/recipes/577784-line-based-side-by-side-diff/
-    """
-
-    if not old_text:
-        old_text = ''
-
-    if not new_text:
-        new_text = ''
-
-    old_text = strip_tags(
-        strip_line_breaks(str(old_text).encode('utf-8', errors='ignore'))
-    )
-    new_text = strip_tags(
-        strip_line_breaks(str(new_text).encode('utf-8', errors='ignore'))
-    )
-
-    old_text = unicodedata.normalize("NFKD", old_text)
-    new_text = unicodedata.normalize("NFKD", new_text)
-
-    def yield_open_entry(open_entry):
-        """ Yield all open changes. """
-        ls, rs = open_entry
-        # Get unchanged parts onto the right line
-        if ls[0] == rs[0]:
-            yield (False, ls[0], rs[0])
-            for l, r in zip_longest(ls[1:], rs[1:]):
-                yield (True, l, r)
-        elif ls[-1] == rs[-1]:
-            for l, r in zip_longest(ls[:-1], rs[:-1]):
-                yield (l != r, l, r)
-            yield (False, ls[-1], rs[-1])
-        else:
-            for l, r in zip_longest(ls, rs):
-                yield (True, l, r)
-
-    line_split = re.compile(r'(?:\r?\n)')
-    dmp = diff_match_patch.diff_match_patch()
-
-    diff = dmp.diff_main(old_text, new_text)
-    dmp.diff_cleanupSemantic(diff)
-
-    open_entry = ([None], [None])
-    for change_type, entry in diff:
-        assert change_type in [-1, 0, 1]
-
-        entry = entry.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
-
-        lines = line_split.split(entry)
-
-        # Merge with previous entry if still open
-        ls, rs = open_entry
-
-        line = lines[0]
-        if line:
-            if change_type == 0:
-                ls[-1] = ls[-1] or ''
-                rs[-1] = rs[-1] or ''
-                ls[-1] += line
-                rs[-1] += line
-            elif change_type == 1:
-                rs[-1] = rs[-1] or ''
-                rs[-1] += '<ins>%s</ins>' % line if line else ''
-            elif change_type == -1:
-                ls[-1] = ls[-1] or ''
-                ls[-1] += '<del>%s</del>' % line if line else ''
-
-        lines = lines[1:]
-
-        if lines:
-            if change_type == 0:
-                # Push out open entry
-                for entry in yield_open_entry(open_entry):
-                    yield entry
-
-                # Directly push out lines until last
-                for line in lines[:-1]:
-                    yield (False, line, line)
-
-                # Keep last line open
-                open_entry = ([lines[-1]], [lines[-1]])
-            elif change_type == 1:
-                ls, rs = open_entry
-
-                for line in lines:
-                    rs.append('<ins>%s</ins>' % line if line else '')
-
-                open_entry = (ls, rs)
-            elif change_type == -1:
-                ls, rs = open_entry
-
-                for line in lines:
-                    ls.append('<del>%s</del>' % line if line else '')
-
-                open_entry = (ls, rs)
-
-    # Push out open entry
-    for entry in yield_open_entry(open_entry):
-        yield entry

app/models/track.py

@@ -57,11 +57,7 @@ def font_color(self):
             h = h.split(',')
             a = (
                 1
-                - (
-                    0.299 * int(int(h[0]), 16)
-                    + 0.587 * int(int(h[1]), 16)
-                    + 0.114 * int(int(h[2]), 16)
-                )
+                - (0.299 * int(h[0], 16) + 0.587 * int(h[1], 16) + 0.114 * int(h[2], 16))
                 / 255
             )
         return '#000000' if (a < 0.5) else '#ffffff'

app/models/user.py

@@ -217,7 +217,6 @@ def is_owner(self, event_id):
         return self._is_role(OWNER, event_id)
 
     def is_organizer(self, event_id):
-        # type: (object) -> object
         return self._is_role(ORGANIZER, event_id)
 
     def is_coorganizer(self, event_id):
@@ -248,7 +247,6 @@ def is_user_owner(self):
 
     @hybrid_property
     def is_user_organizer(self):
-        # type: (object) -> object
         return self._is_role(ORGANIZER)
 
     @hybrid_property