Refactorings by https://sourcery.ai/, revised by @snarfed

originally from #245. thanks @sourcery-ai!

api.py

@@ -61,7 +61,7 @@
 MAX_PATH_LEN = len(PATH_DEFAULTS) + 1
 
 
-@app.app.route(f'/<path:path>', methods=('GET', 'HEAD'))
+@app.app.route('/<path:path>', methods=('GET', 'HEAD'))
 @flask_util.cached(app.cache, app.RESPONSE_CACHE_TIME)
 def api(path):
   """Handles an API GET.

granary/facebook.py

@@ -647,7 +647,7 @@ def _create(self, obj, preview=None, include_link=source.OMIT_LINK,
     content = self._content_for_create(obj, ignore_formatting=ignore_formatting,
                                        strip_first_video_tag=bool(video_url))
 
-    if not content and not (video_url or image_url):
+    if not content and not video_url and not image_url:
       if type == 'activity':
         content = verb
       else:
@@ -776,41 +776,40 @@ def _create(self, obj, preview=None, include_link=source.OMIT_LINK,
       if preview:
         return source.creation_result(content=preview_content,
                                       description='<span class="verb">post</span>:')
+      if video_url:
+        api_call = API_UPLOAD_VIDEO
+        msg_data.update({
+          'file_url': video_url,
+          'description': msg_data.pop('message', ''),
+        })
+      elif image_url:
+        api_call = API_PUBLISH_PHOTO
+        msg_data['url'] = image_url
+        # use Timeline Photos album, if we can find it, since it keeps photo
+        # posts separate instead of consolidating them into a single "X added
+        # n new photos..." post.
+        # https://github.com/snarfed/bridgy/issues/571
+        for album in self.urlopen(API_ALBUMS % 'me', _as=list):
+          id = album.get('id')
+          if id and album.get('type') == 'wall':
+            api_call = API_PUBLISH_ALBUM_PHOTO % id
+            break
+        if people:
+          # tags is JSON list of dicts with tag_uid fields
+          # https://developers.facebook.com/docs/graph-api/reference/user/photos#Creating
+          msg_data['tags'] = json_dumps([{'tag_uid': tag['id']} for tag in people])
       else:
-        if video_url:
-          api_call = API_UPLOAD_VIDEO
-          msg_data.update({
-            'file_url': video_url,
-            'description': msg_data.pop('message', ''),
-          })
-        elif image_url:
-          api_call = API_PUBLISH_PHOTO
-          msg_data['url'] = image_url
-          # use Timeline Photos album, if we can find it, since it keeps photo
-          # posts separate instead of consolidating them into a single "X added
-          # n new photos..." post.
-          # https://github.com/snarfed/bridgy/issues/571
-          for album in self.urlopen(API_ALBUMS % 'me', _as=list):
-            id = album.get('id')
-            if id and album.get('type') == 'wall':
-              api_call = API_PUBLISH_ALBUM_PHOTO % id
-              break
-          if people:
-            # tags is JSON list of dicts with tag_uid fields
-            # https://developers.facebook.com/docs/graph-api/reference/user/photos#Creating
-            msg_data['tags'] = json_dumps([{'tag_uid': tag['id']} for tag in people])
-        else:
-          api_call = API_PUBLISH_POST
-          if people:
-            # tags is comma-separated user id string
-            # https://developers.facebook.com/docs/graph-api/reference/user/feed#pubfields
-            msg_data['tags'] = ','.join(tag['id'] for tag in people)
-
-        resp = self.urlopen(api_call, data=urllib.parse.urlencode(msg_data))
-        resp.update({'url': self.post_url(resp), 'type': 'post'})
-        if video_url and not resp.get('success', True):
-          msg = 'Video upload failed.'
-          return source.creation_result(abort=True, error_plain=msg, error_html=msg)
+        api_call = API_PUBLISH_POST
+        if people:
+          # tags is comma-separated user id string
+          # https://developers.facebook.com/docs/graph-api/reference/user/feed#pubfields
+          msg_data['tags'] = ','.join(tag['id'] for tag in people)
+
+      resp = self.urlopen(api_call, data=urllib.parse.urlencode(msg_data))
+      resp.update({'url': self.post_url(resp), 'type': 'post'})
+      if video_url and not resp.get('success', True):
+        msg = 'Video upload failed.'
+        return source.creation_result(abort=True, error_plain=msg, error_html=msg)
 
     elif type == 'activity' and verb == 'share':
       return source.creation_result(
@@ -1210,7 +1209,7 @@ def post_to_object(self, post, type=None):
     # is there an attachment? prefer to represent it as a picture (ie image
     # object), but if not, fall back to a link.
     att = {
-      'url': link if link else url,
+      'url': link or url,
       'image': {'url': picture},
       'displayName': post.get('name'),
       'summary': post.get('caption'),

granary/flickr.py

@@ -424,12 +424,13 @@ def get_activities_response(self, user_id=None, group_id=None, app_id=None,
       # gives all recent comments and faves, instead of hitting the API for
       # each photo
       if fetch_replies:
-        replies = []
         comments_resp = self.call_api_method('flickr.photos.comments.getList', {
           'photo_id': photo.get('id'),
         })
-        for comment in comments_resp.get('comments', {}).get('comment', []):
-          replies.append(self.comment_to_object(comment, photo.get('id')))
+        replies = [
+            self.comment_to_object(comment, photo.get('id'))
+            for comment in comments_resp.get('comments', {}).get('comment', [])
+        ]
         activity['object']['replies'] = {
           'items': replies,
           'totalItems': len(replies),

granary/github.py

@@ -343,11 +343,8 @@ def get_activities_response(self, user_id=None, group_id=None, app_id=None,
     if fetch_shares or fetch_events or fetch_mentions or search_query:
       raise NotImplementedError()
 
-    since = None
     etag_parsed = email.utils.parsedate(etag)
-    if etag_parsed:
-      since = datetime.datetime(*etag_parsed[:6])
-
+    since = datetime.datetime(*etag_parsed[:6]) if etag_parsed else None
     activities = []
 
     if activity_id:
@@ -654,14 +651,14 @@ def _create(self, obj, preview=None, include_link=source.OMIT_LINK,
         return source.creation_result(
           description='<span class="verb">star</span> <a href="%s">%s/%s</a>.' %
             (base_url, owner, repo))
-      else:
-        issue = self.graphql(GRAPHQL_REPO, locals())
-        resp = self.graphql(GRAPHQL_ADD_STAR, {
-          'starrable_id': issue['repository']['id'],
-        })
-        return source.creation_result({
-          'url': base_url + '/stargazers',
-        })
+
+      issue = self.graphql(GRAPHQL_REPO, locals())
+      resp = self.graphql(GRAPHQL_ADD_STAR, {
+        'starrable_id': issue['repository']['id'],
+      })
+      return source.creation_result({
+        'url': base_url + '/stargazers',
+      })
 
     elif type == 'tag':  # add label
       if not (len(path) == 4 and path[2] in ('issues', 'pull')):
@@ -687,13 +684,13 @@ def _create(self, obj, preview=None, include_link=source.OMIT_LINK,
         return source.creation_result(
           description='add label%s <span class="verb">%s</span> to %s.' % (
             ('s' if len(labels) > 1 else ''), ', '.join(labels), issue_link))
-      else:
-        resp = self.rest(REST_API_ISSUE_LABELS % (owner, repo, number), labels)
-        return source.creation_result({
-          'url': base_url,
-          'type': 'tag',
-          'tags': labels,
-        })
+
+      resp = self.rest(REST_API_ISSUE_LABELS % (owner, repo, number), labels)
+      return source.creation_result({
+        'url': base_url,
+        'type': 'tag',
+        'tags': labels,
+      })
 
     else:  # new issue
       if not (len(path) == 2 or (len(path) == 3 and path[2] == 'issues')):
@@ -744,7 +741,7 @@ def existing_labels(self, owner, repo):
     if not repo:
       return set()
 
-    return set(node['name'] for node in repo['labels']['nodes'])
+    return {node['name'] for node in repo['labels']['nodes']}
 
   def issue_to_object(self, issue):
     """Converts a GitHub issue or pull request to ActivityStreams.

granary/instagram.py

@@ -135,15 +135,15 @@ def get_actor(self, user_id=None, **kwargs):
         return {}
       user_id = 'self'
 
-    if self.scrape:
-      resp = self.get_activities_response(
-        group_id=source.SELF, user_id=user_id, **kwargs)
-      items = resp.get('items')
-      return items[0].get('actor') if items else {}
-    else:
+    if not self.scrape:
       return self.user_to_actor(util.trim_nulls(
         self.urlopen(API_USER_URL % user_id) or {}))
 
+    resp = self.get_activities_response(
+      group_id=source.SELF, user_id=user_id, **kwargs)
+    items = resp.get('items')
+    return items[0].get('actor') if items else {}
+
   def get_activities_response(self, user_id=None, group_id=None, app_id=None,
                               activity_id=None, start_index=0, count=0,
                               etag=None, min_id=None, cache=None,
@@ -884,11 +884,8 @@ def scraped_to_activities(self, html, cookie=None, count=None,
 
       activities.append(util.trim_nulls(activity))
 
-    actor = None
     user = self._json_user_to_user(viewer_user or profile_user)
-    if user:
-      actor = self.user_to_actor(user)
-
+    actor = self.user_to_actor(user) if user else None
     return activities, actor
 
   html_to_activities = scraped_to_activities

granary/mastodon.py

@@ -343,7 +343,7 @@ def status_to_object(self, status):
     }
 
     reblog = status.get('reblog')
-    base_status = reblog if reblog else status
+    base_status = reblog or status
 
     # media! into attachments.
     for media in status.get('media_attachments', []):

granary/meetup.py

@@ -40,7 +40,6 @@ def embed_post(cls, obj):
 
   def __init__(self, access_token):
     self.access_token = access_token
-    pass
 
   def create(self, obj, include_link=source.OMIT_LINK, ignore_formatting=False):
     return self._create(obj, False, include_link, ignore_formatting)
@@ -72,7 +71,7 @@ def _create(self, obj, preview=False, include_link=source.OMIT_LINK, ignore_form
       response = 'yes'
     elif verb == 'rsvp-no':
       response = 'no'
-    elif verb == 'rsvp-maybe' or verb == 'rsvp-interested':
+    elif verb in ('rsvp-maybe', 'rsvp-interested'):
       return self.return_error('Meetup.com does not support %(verb)s' % {'verb': verb})
     else:
       return self.return_error('Meetup.com syndication does not support %(verb)s' % {'verb': verb})

granary/microformats2.py

@@ -227,10 +227,8 @@ def object_to_json(obj, trim_nulls=True, entry_class='h-entry',
       logging('Ignoring duration %r; expected int, got %s', duration.__class__)
       duration = None
 
-  sizes = []
   size = stream.get('size')
-  if size:
-    sizes = [str(size)]
+  sizes = [str(size)] if size else []
 
   # construct mf2!
   ret = {

granary/reddit.py

@@ -242,10 +242,8 @@ def _fetch_replies(self, r, activities):
       # for v0 we will use just the top level comments because threading is hard.
       # feature request: https://github.com/snarfed/bridgy/issues/1014
       subm.comments.replace_more()
-      replies = []
-      for top_level_comment in subm.comments:
-        replies.append(self.praw_to_activity(top_level_comment, 'comment'))
-
+      replies = [self.praw_to_activity(top_level_comment, 'comment')
+                 for top_level_comment in subm.comments]
       items = [r.get('object') for r in replies]
       activity['object']['replies'] = {
         'items': items,

granary/source.py

@@ -662,7 +662,7 @@ def postprocess_activity(activity):
           activity['title'] = obj_name
         elif verb and (obj_name or obj_type):
           app = activity.get('generator', {}).get('displayName')
-          name = obj_name if obj_name else 'a %s' % (obj_type or 'unknown')
+          name = obj_name or 'a %s' % (obj_type or 'unknown')
           app = ' on %s' % app if app else ''
           activity['title'] = '%s %s %s%s.' % (actor_name, verb or 'posted',
                                                name, app)
@@ -767,10 +767,10 @@ def original_post_discovery(
       if not domain:
         continue
 
-      if not include_reserved_hosts:
-        if ('.' not in domain or
-            domain.split('.')[-1] in (util.RESERVED_TLDS | util.LOCAL_TLDS)):
-          continue
+      if not include_reserved_hosts and (
+          ('.' not in domain
+           or domain.split('.')[-1] in (util.RESERVED_TLDS | util.LOCAL_TLDS))):
+        continue
 
       which = (originals if not domains or util.domain_or_parent_in(domain, domains)
                else mentions)

granary/tests/test_facebook.py

@@ -2900,7 +2900,7 @@ def test_create_interested_rsvp_error(self):
 
   def test_create_rsvp_multi_instance_event_error(self):
     """Facebook API doesn't allow RSVPing to multi-instance aka recurring events."""
-    for i in range(2):
+    for _ in range(2):
       self.expect_urlopen(API_EVENT % '234', MULTI_EVENT)
     self.mox.ReplayAll()
 
@@ -3157,7 +3157,7 @@ def check(expected, id, is_comment):
       check(expected, id, True)
 
   def test_resolve_object_id(self):
-    for i in range(3):
+    for _ in range(3):
       self.expect_urlopen(API_OBJECT % ('111', '222'),
                           {'id': '0', 'object_id': '333'})
     self.mox.ReplayAll()

granary/tests/test_github.py

@@ -201,7 +201,6 @@ def tag_uri(name):
   'user': USER_REST,
   'title': 'a PR to merge',
   'body': 'a PR message',
-  'comments_url': 'https://api.github.com/repos/foo/bar/issues/444/comments',
   'issue_url': 'https://api.github.com/repos/foo/bar/issues/444',
   'diff_url': 'https://github.com/foo/bar/pull/444.diff',
   'patch_url': 'https://github.com/foo/bar/pull/444.patch',
@@ -274,7 +273,6 @@ def tag_uri(name):
   'url': 'https://github.com/foo/bar/pull/123#issuecomment-456',
   'author': ACTOR,
   'content': 'i have something to say here',
-  'published': '2012-12-05T00:58:26+00:00',
   'inReplyTo': [{'url': 'https://github.com/foo/bar/pull/123'}],
   'published': '2015-07-23T18:47:58+00:00',
   'updated': '2015-07-23T19:47:58+00:00',
@@ -869,7 +867,7 @@ def test_create_blockquote(self):
     self.assertIsNone(result.error_plain, result)
 
   def test_preview_issue(self):
-    for i in range(2):
+    for _ in range(2):
       self.expect_graphql_get_labels(['new silo'])
       rendered = self.expect_markdown_render(ISSUE_OBJ['content'].strip())
     self.mox.ReplayAll()

granary/tests/test_instagram.py

@@ -1167,7 +1167,7 @@ def test_get_activities_scrape_fetch_extras_cache(self):
     self.mox.ReplayAll()
 
     cache = util.CacheDict()
-    for i in range(3):
+    for _ in range(3):
       self.instagram.get_activities(user_id='x', group_id=source.SELF,
                                     fetch_likes=True, fetch_replies=True,
                                     scrape=True, cache=cache, cookie='kuky')
@@ -1288,7 +1288,7 @@ def test_get_activities_scrape_rate_limited(self):
     # first attempt: rate limited
     self.expect_requests_get(HTML_BASE_URL, status_code=429, cookie='kuky')
     # third attempt ignore rate limit lock. fourth limit is past lock.
-    for i in range(2):
+    for _ in range(2):
       self.expect_requests_get('x/', HTML_PROFILE_COMPLETE)
     self.mox.ReplayAll()
 

granary/tests/test_mastodon.py

@@ -502,7 +502,7 @@ def test_get_activities_fetch_cache(self):
 
     self.mox.ReplayAll()
     cache = util.CacheDict()
-    for i in range(4):
+    for _ in range(4):
       self.mastodon.get_activities(fetch_replies=True, fetch_shares=True,
                                    fetch_likes=True, cache=cache)