Add type hints to schema deltas

changelog.d/15496.misc

@@ -0,0 +1 @@
+Improve type hints.

mypy.ini

@@ -28,13 +28,10 @@ files =
 # https://docs.python.org/3/library/re.html#re.X
 exclude = (?x)
   ^(
-   |synapse/storage/databases/__init__.py
-   |synapse/storage/databases/main/cache.py
    |synapse/storage/schema/
    )$
 
 [mypy-synapse.metrics._reactor_metrics]
-disallow_untyped_defs = False
 # This module imports select.epoll. That exists on Linux, but doesn't on macOS.
 # See https://github.com/matrix-org/synapse/pull/11771.
 warn_unused_ignores = False

synapse/storage/databases/__init__.py

@@ -95,7 +95,7 @@ def __init__(self, main_store_class: Type[DataStoreT], hs: "HomeServer"):
                     # If we're on a process that can persist events also
                     # instantiate a `PersistEventsStore`
                     if hs.get_instance_name() in hs.config.worker.writers.events:
-                        persist_events = PersistEventsStore(hs, database, main, db_conn)
+                        persist_events = PersistEventsStore(hs, database, main, db_conn)  # type: ignore[arg-type]
 
                 if "state" in database_config.databases:
                     logger.info(
@@ -133,6 +133,6 @@ def __init__(self, main_store_class: Type[DataStoreT], hs: "HomeServer"):
 
         # We use local variables here to ensure that the databases do not have
         # optional types.
-        self.main = main
+        self.main = main  # type: ignore[assignment]
         self.state = state
         self.persist_events = persist_events

synapse/storage/databases/main/cache.py

@@ -205,13 +205,13 @@ def _process_event_stream_row(self, token: int, row: EventsStreamRow) -> None:
             )
         elif row.type == EventsStreamCurrentStateRow.TypeId:
             assert isinstance(data, EventsStreamCurrentStateRow)
-            self._curr_state_delta_stream_cache.entity_has_changed(data.room_id, token)
+            self._curr_state_delta_stream_cache.entity_has_changed(data.room_id, token)  # type: ignore[attr-defined]
 
             if data.type == EventTypes.Member:
-                self.get_rooms_for_user_with_stream_ordering.invalidate(
+                self.get_rooms_for_user_with_stream_ordering.invalidate(  # type: ignore[attr-defined]
                     (data.state_key,)
                 )
-                self.get_rooms_for_user.invalidate((data.state_key,))
+                self.get_rooms_for_user.invalidate((data.state_key,))  # type: ignore[attr-defined]
         else:
             raise Exception("Unknown events stream row type %s" % (row.type,))
 
@@ -229,7 +229,7 @@ def _invalidate_caches_for_event(
         # This invalidates any local in-memory cached event objects, the original
         # process triggering the invalidation is responsible for clearing any external
         # cached objects.
-        self._invalidate_local_get_event_cache(event_id)
+        self._invalidate_local_get_event_cache(event_id)  # type: ignore[attr-defined]
 
         self._attempt_to_invalidate_cache("have_seen_event", (room_id, event_id))
         self._attempt_to_invalidate_cache("get_latest_event_ids_in_room", (room_id,))
@@ -242,10 +242,10 @@ def _invalidate_caches_for_event(
         self._attempt_to_invalidate_cache("_get_membership_from_event_id", (event_id,))
 
         if not backfilled:
-            self._events_stream_cache.entity_has_changed(room_id, stream_ordering)
+            self._events_stream_cache.entity_has_changed(room_id, stream_ordering)  # type: ignore[attr-defined]
 
         if redacts:
-            self._invalidate_local_get_event_cache(redacts)
+            self._invalidate_local_get_event_cache(redacts)  # type: ignore[attr-defined]
             # Caches which might leak edits must be invalidated for the event being
             # redacted.
             self._attempt_to_invalidate_cache("get_relations_for_event", (redacts,))
@@ -254,7 +254,7 @@ def _invalidate_caches_for_event(
             self._attempt_to_invalidate_cache("get_thread_id_for_receipts", (redacts,))
 
         if etype == EventTypes.Member:
-            self._membership_stream_cache.entity_has_changed(state_key, stream_ordering)
+            self._membership_stream_cache.entity_has_changed(state_key, stream_ordering)  # type: ignore[attr-defined]
             self._attempt_to_invalidate_cache(
                 "get_invited_rooms_for_local_user", (state_key,)
             )
@@ -378,6 +378,8 @@ def _send_invalidation_to_replication(
             )
 
         if isinstance(self.database_engine, PostgresEngine):
+            assert self._cache_id_gen is not None
+
             # get_next() returns a context manager which is designed to wrap
             # the transaction. However, we want to only get an ID when we want
             # to use it, here, so we need to call __enter__ manually, and have

synapse/storage/schema/main/delta/20/pushers.py

@@ -81,7 +81,3 @@ def run_create(cur, database_engine, *args, **kwargs):
     cur.execute("DROP TABLE pushers")
     cur.execute("ALTER TABLE pushers2 RENAME TO pushers")
     logger.info("Moved %d pushers to new table", count)
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/25/fts.py

@@ -72,7 +72,3 @@ def run_create(cur, database_engine, *args, **kwargs):
         )
 
         cur.execute(sql, ("event_search", progress_json))
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/27/ts.py

@@ -51,7 +51,3 @@ def run_create(cur, database_engine, *args, **kwargs):
         )
 
         cur.execute(sql, ("event_origin_server_ts", progress_json))
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/31/pushers.py → synapse/storage/schema/main/delta/31/pushers_0.py

@@ -80,7 +80,3 @@ def run_create(cur, database_engine, *args, **kwargs):
     cur.execute("DROP TABLE pushers")
     cur.execute("ALTER TABLE pushers2 RENAME TO pushers")
     logger.info("Moved %d pushers to new table", count)
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/31/search_update.py

@@ -56,7 +56,3 @@ def run_create(cur, database_engine, *args, **kwargs):
         )
 
         cur.execute(sql, ("event_search_order", progress_json))
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/33/event_fields.py

@@ -51,7 +51,3 @@ def run_create(cur, database_engine, *args, **kwargs):
         )
 
         cur.execute(sql, ("event_fields_sender_url", progress_json))
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/34/cache_stream.py

@@ -40,7 +40,3 @@ def run_create(cur, database_engine, *args, **kwargs):
 
     for statement in get_statements(CREATE_TABLE.splitlines()):
         cur.execute(statement)
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/34/received_txn_purge.py

@@ -26,7 +26,3 @@ def run_create(cur, database_engine, *args, **kwargs):
         cur.execute("DELETE FROM received_transactions")
 
     cur.execute("CREATE INDEX received_transactions_ts ON received_transactions(ts)")
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/37/remove_auth_idx.py

@@ -79,7 +79,3 @@ def run_create(cur, database_engine, *args, **kwargs):
 
     for statement in get_statements(drop_constraint.splitlines()):
         cur.execute(statement)
-
-
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass

synapse/storage/schema/main/delta/42/user_dir.py

@@ -78,7 +78,3 @@ def run_create(cur, database_engine, *args, **kwargs):
             cur.execute(statement)
     else:
         raise Exception("Unrecognized database engine")
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/48/group_unique_indexes.py

@@ -57,7 +57,3 @@ def run_create(cur, database_engine, *args, **kwargs):
 
     for statement in get_statements(FIX_INDEXES.splitlines()):
         cur.execute(statement)
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/56/unique_user_filter_index.py

@@ -16,10 +16,6 @@
 """
 
 
-def run_upgrade(cur, database_engine, *args, **kwargs):
-    pass
-
-
 def run_create(cur, database_engine, *args, **kwargs):
     if isinstance(database_engine, PostgresEngine):
         select_clause = """

synapse/storage/schema/main/delta/58/06dlols_unique_idx.py

@@ -27,10 +27,6 @@
 logger = logging.getLogger(__name__)
 
 
-def run_upgrade(*args, **kwargs):
-    pass
-
-
 def run_create(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs):
     # some instances might already have this index, in which case we can skip this
     if isinstance(database_engine, PostgresEngine):

synapse/storage/schema/main/delta/58/11user_id_seq.py

@@ -28,7 +28,3 @@ def run_create(cur, database_engine, *args, **kwargs):
 
     next_id = find_max_generated_user_id_localpart(cur) + 1
     cur.execute("CREATE SEQUENCE user_id_seq START WITH %s", (next_id,))
-
-
-def run_upgrade(*args, **kwargs):
-    pass

synapse/storage/schema/main/delta/59/01ignored_user.py

@@ -27,10 +27,6 @@
 logger = logging.getLogger(__name__)
 
 
-def run_upgrade(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs):
-    pass
-
-
 def run_create(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs):
     logger.info("Creating ignored_users table")
     execute_statements_from_stream(cur, StringIO(_create_commands))

synapse/storage/schema/main/delta/61/03recreate_min_depth.py

@@ -64,7 +64,3 @@ def run_create(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs
             (6103, 'replace_room_depth_min_depth', '{}', 'populate_room_depth2')
         """
     )
-
-
-def run_upgrade(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs):
-    pass

synapse/storage/schema/state/delta/47/state_group_seq.py

@@ -28,7 +28,3 @@ def run_create(cur, database_engine, *args, **kwargs):
             start_val = row[0] + 1
 
         cur.execute("CREATE SEQUENCE state_group_id_seq START WITH %s", (start_val,))
-
-
-def run_upgrade(*args, **kwargs):
-    pass