Run black on the rest of the storage module (#4996)

1 files changed, 94 insertions, 91 deletions

diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py
index 13ce296a9b..a729f3e067 100644
--- a/synapse/storage/event_push_actions.py
+++ b/synapse/storage/event_push_actions.py
@@ -31,7 +31,9 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_NOTIF_ACTION = ["notify", {"set_tweak": "highlight", "value": False}]
 DEFAULT_HIGHLIGHT_ACTION = [
-    "notify", {"set_tweak": "sound", "value": "default"}, {"set_tweak": "highlight"}
+    "notify",
+    {"set_tweak": "sound", "value": "default"},
+    {"set_tweak": "highlight"},
 ]
 
 
@@ -91,25 +93,26 @@ class EventPushActionsWorkerStore(SQLBaseStore):
 
     @cachedInlineCallbacks(num_args=3, tree=True, max_entries=5000)
     def get_unread_event_push_actions_by_room_for_user(
-            self, room_id, user_id, last_read_event_id
+        self, room_id, user_id, last_read_event_id
     ):
         ret = yield self.runInteraction(
             "get_unread_event_push_actions_by_room",
             self._get_unread_counts_by_receipt_txn,
-            room_id, user_id, last_read_event_id
+            room_id,
+            user_id,
+            last_read_event_id,
         )
         defer.returnValue(ret)
 
-    def _get_unread_counts_by_receipt_txn(self, txn, room_id, user_id,
-                                          last_read_event_id):
+    def _get_unread_counts_by_receipt_txn(
+        self, txn, room_id, user_id, last_read_event_id
+    ):
         sql = (
             "SELECT stream_ordering"
             " FROM events"
             " WHERE room_id = ? AND event_id = ?"
         )
-        txn.execute(
-            sql, (room_id, last_read_event_id)
-        )
+        txn.execute(sql, (room_id, last_read_event_id))
         results = txn.fetchall()
         if len(results) == 0:
             return {"notify_count": 0, "highlight_count": 0}
@@ -138,10 +141,13 @@ class EventPushActionsWorkerStore(SQLBaseStore):
         row = txn.fetchone()
         notify_count = row[0] if row else 0
 
-        txn.execute("""
+        txn.execute(
+            """
             SELECT notif_count FROM event_push_summary
             WHERE room_id = ? AND user_id = ? AND stream_ordering > ?
-        """, (room_id, user_id, stream_ordering,))
+        """,
+            (room_id, user_id, stream_ordering),
+        )
         rows = txn.fetchall()
         if rows:
             notify_count += rows[0][0]
@@ -161,10 +167,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
         row = txn.fetchone()
         highlight_count = row[0] if row else 0
 
-        return {
-            "notify_count": notify_count,
-            "highlight_count": highlight_count,
-        }
+        return {"notify_count": notify_count, "highlight_count": highlight_count}
 
     @defer.inlineCallbacks
     def get_push_action_users_in_range(self, min_stream_ordering, max_stream_ordering):
@@ -175,6 +178,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             )
             txn.execute(sql, (min_stream_ordering, max_stream_ordering))
             return [r[0] for r in txn]
+
         ret = yield self.runInteraction("get_push_action_users_in_range", f)
         defer.returnValue(ret)
 
@@ -223,12 +227,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "   AND ep.stream_ordering <= ?"
                 " ORDER BY ep.stream_ordering ASC LIMIT ?"
             )
-            args = [
-                user_id, user_id,
-                min_stream_ordering, max_stream_ordering, limit,
-            ]
+            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return txn.fetchall()
+
         after_read_receipt = yield self.runInteraction(
             "get_unread_push_actions_for_user_in_range_http_arr", get_after_receipt
         )
@@ -253,12 +255,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "   AND ep.stream_ordering <= ?"
                 " ORDER BY ep.stream_ordering ASC LIMIT ?"
             )
-            args = [
-                user_id, user_id,
-                min_stream_ordering, max_stream_ordering, limit,
-            ]
+            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return txn.fetchall()
+
         no_read_receipt = yield self.runInteraction(
             "get_unread_push_actions_for_user_in_range_http_nrr", get_no_receipt
         )
@@ -269,7 +269,8 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "room_id": row[1],
                 "stream_ordering": row[2],
                 "actions": _deserialize_action(row[3], row[4]),
-            } for row in after_read_receipt + no_read_receipt
+            }
+            for row in after_read_receipt + no_read_receipt
         ]
 
         # Now sort it so it's ordered correctly, since currently it will
@@ -326,12 +327,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "   AND ep.stream_ordering <= ?"
                 " ORDER BY ep.stream_ordering DESC LIMIT ?"
             )
-            args = [
-                user_id, user_id,
-                min_stream_ordering, max_stream_ordering, limit,
-            ]
+            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return txn.fetchall()
+
         after_read_receipt = yield self.runInteraction(
             "get_unread_push_actions_for_user_in_range_email_arr", get_after_receipt
         )
@@ -356,12 +355,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "   AND ep.stream_ordering <= ?"
                 " ORDER BY ep.stream_ordering DESC LIMIT ?"
             )
-            args = [
-                user_id, user_id,
-                min_stream_ordering, max_stream_ordering, limit,
-            ]
+            args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
             txn.execute(sql, args)
             return txn.fetchall()
+
         no_read_receipt = yield self.runInteraction(
             "get_unread_push_actions_for_user_in_range_email_nrr", get_no_receipt
         )
@@ -374,7 +371,8 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "stream_ordering": row[2],
                 "actions": _deserialize_action(row[3], row[4]),
                 "received_ts": row[5],
-            } for row in after_read_receipt + no_read_receipt
+            }
+            for row in after_read_receipt + no_read_receipt
         ]
 
         # Now sort it so it's ordered correctly, since currently it will
@@ -408,7 +406,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 LIMIT 1
             """
 
-            txn.execute(sql, (user_id, min_stream_ordering,))
+            txn.execute(sql, (user_id, min_stream_ordering))
             return bool(txn.fetchone())
 
         return self.runInteraction(
@@ -454,10 +452,13 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 VALUES (?, ?, ?, ?, ?)
             """
 
-            txn.executemany(sql, (
-                _gen_entry(user_id, actions)
-                for user_id, actions in iteritems(user_id_actions)
-            ))
+            txn.executemany(
+                sql,
+                (
+                    _gen_entry(user_id, actions)
+                    for user_id, actions in iteritems(user_id_actions)
+                ),
+            )
 
         return self.runInteraction(
             "add_push_actions_to_staging", _add_push_actions_to_staging_txn
@@ -475,9 +476,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
         try:
             res = yield self._simple_delete(
                 table="event_push_actions_staging",
-                keyvalues={
-                    "event_id": event_id,
-                },
+                keyvalues={"event_id": event_id},
                 desc="remove_push_actions_from_staging",
             )
             defer.returnValue(res)
@@ -486,7 +485,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             # another exception here really isn't helpful - there's nothing
             # the caller can do about it. Just log the exception and move on.
             logger.exception(
-                "Error removing push actions after event persistence failure",
+                "Error removing push actions after event persistence failure"
             )
 
     def _find_stream_orderings_for_times(self):
@@ -503,16 +502,14 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             txn, self._clock.time_msec() - 30 * 24 * 60 * 60 * 1000
         )
         logger.info(
-            "Found stream ordering 1 month ago: it's %d",
-            self.stream_ordering_month_ago
+            "Found stream ordering 1 month ago: it's %d", self.stream_ordering_month_ago
         )
         logger.info("Searching for stream ordering 1 day ago")
         self.stream_ordering_day_ago = self._find_first_stream_ordering_after_ts_txn(
             txn, self._clock.time_msec() - 24 * 60 * 60 * 1000
         )
         logger.info(
-            "Found stream ordering 1 day ago: it's %d",
-            self.stream_ordering_day_ago
+            "Found stream ordering 1 day ago: it's %d", self.stream_ordering_day_ago
         )
 
     def find_first_stream_ordering_after_ts(self, ts):
@@ -631,16 +628,17 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             index_name="event_push_actions_highlights_index",
             table="event_push_actions",
             columns=["user_id", "room_id", "topological_ordering", "stream_ordering"],
-            where_clause="highlight=1"
+            where_clause="highlight=1",
         )
 
         self._doing_notif_rotation = False
         self._rotate_notif_loop = self._clock.looping_call(
-            self._start_rotate_notifs, 30 * 60 * 1000,
+            self._start_rotate_notifs, 30 * 60 * 1000
         )
 
-    def _set_push_actions_for_event_and_users_txn(self, txn, events_and_contexts,
-                                                  all_events_and_contexts):
+    def _set_push_actions_for_event_and_users_txn(
+        self, txn, events_and_contexts, all_events_and_contexts
+    ):
         """Handles moving push actions from staging table to main
         event_push_actions table for all events in `events_and_contexts`.
 
@@ -667,43 +665,44 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         """
 
         if events_and_contexts:
-            txn.executemany(sql, (
+            txn.executemany(
+                sql,
                 (
-                    event.room_id, event.internal_metadata.stream_ordering,
-                    event.depth, event.event_id,
-                )
-                for event, _ in events_and_contexts
-            ))
+                    (
+                        event.room_id,
+                        event.internal_metadata.stream_ordering,
+                        event.depth,
+                        event.event_id,
+                    )
+                    for event, _ in events_and_contexts
+                ),
+            )
 
         for event, _ in events_and_contexts:
             user_ids = self._simple_select_onecol_txn(
                 txn,
                 table="event_push_actions_staging",
-                keyvalues={
-                    "event_id": event.event_id,
-                },
+                keyvalues={"event_id": event.event_id},
                 retcol="user_id",
             )
 
             for uid in user_ids:
                 txn.call_after(
                     self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
-                    (event.room_id, uid,)
+                    (event.room_id, uid),
                 )
 
         # Now we delete the staging area for *all* events that were being
         # persisted.
         txn.executemany(
             "DELETE FROM event_push_actions_staging WHERE event_id = ?",
-            (
-                (event.event_id,)
-                for event, _ in all_events_and_contexts
-            )
+            ((event.event_id,) for event, _ in all_events_and_contexts),
         )
 
     @defer.inlineCallbacks
-    def get_push_actions_for_user(self, user_id, before=None, limit=50,
-                                  only_highlight=False):
+    def get_push_actions_for_user(
+        self, user_id, before=None, limit=50, only_highlight=False
+    ):
         def f(txn):
             before_clause = ""
             if before:
@@ -727,15 +726,12 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                 " WHERE epa.event_id = e.event_id"
                 " AND epa.user_id = ? %s"
                 " ORDER BY epa.stream_ordering DESC"
-                " LIMIT ?"
-                % (before_clause,)
+                " LIMIT ?" % (before_clause,)
             )
             txn.execute(sql, args)
             return self.cursor_to_dict(txn)
 
-        push_actions = yield self.runInteraction(
-            "get_push_actions_for_user", f
-        )
+        push_actions = yield self.runInteraction("get_push_actions_for_user", f)
         for pa in push_actions:
             pa["actions"] = _deserialize_action(pa["actions"], pa["highlight"])
         defer.returnValue(push_actions)
@@ -753,6 +749,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             )
             txn.execute(sql, (stream_ordering,))
             return txn.fetchone()
+
         result = yield self.runInteraction("get_time_of_last_push_action_before", f)
         defer.returnValue(result[0] if result else None)
 
@@ -761,24 +758,24 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         def f(txn):
             txn.execute("SELECT MAX(stream_ordering) FROM event_push_actions")
             return txn.fetchone()
-        result = yield self.runInteraction(
-            "get_latest_push_action_stream_ordering", f
-        )
+
+        result = yield self.runInteraction("get_latest_push_action_stream_ordering", f)
         defer.returnValue(result[0] or 0)
 
     def _remove_push_actions_for_event_id_txn(self, txn, room_id, event_id):
         # Sad that we have to blow away the cache for the whole room here
         txn.call_after(
             self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
-            (room_id,)
+            (room_id,),
         )
         txn.execute(
             "DELETE FROM event_push_actions WHERE room_id = ? AND event_id = ?",
-            (room_id, event_id)
+            (room_id, event_id),
         )
 
-    def _remove_old_push_actions_before_txn(self, txn, room_id, user_id,
-                                            stream_ordering):
+    def _remove_old_push_actions_before_txn(
+        self, txn, room_id, user_id, stream_ordering
+    ):
         """
         Purges old push actions for a user and room before a given
         stream_ordering.
@@ -795,7 +792,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         """
         txn.call_after(
             self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
-            (room_id, user_id, )
+            (room_id, user_id),
         )
 
         # We need to join on the events table to get the received_ts for
@@ -811,13 +808,16 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             " WHERE user_id = ? AND room_id = ? AND "
             " stream_ordering <= ?"
             " AND ((stream_ordering < ? AND highlight = 1) or highlight = 0)",
-            (user_id, room_id, stream_ordering, self.stream_ordering_month_ago)
+            (user_id, room_id, stream_ordering, self.stream_ordering_month_ago),
         )
 
-        txn.execute("""
+        txn.execute(
+            """
             DELETE FROM event_push_summary
             WHERE room_id = ? AND user_id = ? AND stream_ordering <= ?
-        """, (room_id, user_id, stream_ordering))
+        """,
+            (room_id, user_id, stream_ordering),
+        )
 
     def _start_rotate_notifs(self):
         return run_as_background_process("rotate_notifs", self._rotate_notifs)
@@ -833,8 +833,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                 logger.info("Rotating notifications")
 
                 caught_up = yield self.runInteraction(
-                    "_rotate_notifs",
-                    self._rotate_notifs_txn
+                    "_rotate_notifs", self._rotate_notifs_txn
                 )
                 if caught_up:
                     break
@@ -856,11 +855,14 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
 
         # We don't to try and rotate millions of rows at once, so we cap the
         # maximum stream ordering we'll rotate before.
-        txn.execute("""
+        txn.execute(
+            """
             SELECT stream_ordering FROM event_push_actions
             WHERE stream_ordering > ?
             ORDER BY stream_ordering ASC LIMIT 1 OFFSET ?
-        """, (old_rotate_stream_ordering, self._rotate_count))
+        """,
+            (old_rotate_stream_ordering, self._rotate_count),
+        )
         stream_row = txn.fetchone()
         if stream_row:
             offset_stream_ordering, = stream_row
@@ -904,7 +906,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             LEFT JOIN event_push_summary AS old USING (user_id, room_id)
         """
 
-        txn.execute(sql, (old_rotate_stream_ordering, rotate_to_stream_ordering,))
+        txn.execute(sql, (old_rotate_stream_ordering, rotate_to_stream_ordering))
         rows = txn.fetchall()
 
         logger.info("Rotating notifications, handling %d rows", len(rows))
@@ -922,8 +924,9 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                     "notif_count": row[2],
                     "stream_ordering": row[3],
                 }
-                for row in rows if row[4] is None
-            ]
+                for row in rows
+                if row[4] is None
+            ],
         )
 
         txn.executemany(
@@ -931,20 +934,20 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                 UPDATE event_push_summary SET notif_count = ?, stream_ordering = ?
                 WHERE user_id = ? AND room_id = ?
             """,
-            ((row[2], row[3], row[0], row[1],) for row in rows if row[4] is not None)
+            ((row[2], row[3], row[0], row[1]) for row in rows if row[4] is not None),
         )
 
         txn.execute(
             "DELETE FROM event_push_actions"
             " WHERE ? <= stream_ordering AND stream_ordering < ? AND highlight = 0",
-            (old_rotate_stream_ordering, rotate_to_stream_ordering,)
+            (old_rotate_stream_ordering, rotate_to_stream_ordering),
         )
 
         logger.info("Rotating notifications, deleted %s push actions", txn.rowcount)
 
         txn.execute(
             "UPDATE event_push_summary_stream_ordering SET stream_ordering = ?",
-            (rotate_to_stream_ordering,)
+            (rotate_to_stream_ordering,),
         )