Update type hints and apply other post-blackening cleanup

mautrix_twitter/__main__.py

@@ -1,5 +1,5 @@
 # mautrix-twitter - A Matrix-Twitter DM puppeting bridge
-# Copyright (C) 2021 Tulir Asokan
+# Copyright (C) 2022 Tulir Asokan
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -13,7 +13,9 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import Any, Dict
+from __future__ import annotations
+
+from typing import Any
 
 from mautrix.bridge import Bridge
 from mautrix.types import RoomID, UserID
@@ -94,7 +96,7 @@ def is_bridge_ghost(self, user_id: UserID) -> bool:
     async def count_logged_in_users(self) -> int:
         return len([user for user in User.by_twid.values() if user.twid])
 
-    async def manhole_global_namespace(self, user_id: UserID) -> Dict[str, Any]:
+    async def manhole_global_namespace(self, user_id: UserID) -> dict[str, Any]:
         return {
             **await super().manhole_global_namespace(user_id),
             "User": User,

mautrix_twitter/commands/auth.py

@@ -62,7 +62,7 @@ async def enter_login_cookies(evt: CommandEvent) -> None:
         return
     if len(evt.args) == 0:
         await evt.reply(
-            "Please enter the value of the `ct0` cookie, or use " "the `cancel` command to cancel."
+            "Please enter the value of the `ct0` cookie, or use the `cancel` command to cancel."
         )
         return
 
@@ -83,7 +83,7 @@ async def enter_login_cookies(evt: CommandEvent) -> None:
 @command_handler(
     needs_auth=True,
     help_section=SECTION_AUTH,
-    help_text="Disconnect the bridge from" "your Twitter account",
+    help_text="Disconnect the bridge from your Twitter account",
 )
 async def logout(evt: CommandEvent) -> None:
     await evt.sender.logout()

mautrix_twitter/config.py

@@ -1,5 +1,5 @@
 # mautrix-twitter - A Matrix-Twitter DM puppeting bridge
-# Copyright (C) 2020 Tulir Asokan
+# Copyright (C) 2022 Tulir Asokan
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -13,7 +13,9 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import Any, List, NamedTuple
+from __future__ import annotations
+
+from typing import Any, NamedTuple
 import os
 
 from mautrix.bridge.config import BaseBridgeConfig
@@ -32,7 +34,7 @@ def __getitem__(self, key: str) -> Any:
             return super().__getitem__(key)
 
     @property
-    def forbidden_defaults(self) -> List[ForbiddenDefault]:
+    def forbidden_defaults(self) -> list[ForbiddenDefault]:
         return [
             *super().forbidden_defaults,
             ForbiddenDefault("appservice.database", "postgres://username:password@hostname/db"),
@@ -51,8 +53,6 @@ def do_update(self, helper: ConfigUpdateHelper) -> None:
         if base["appservice.provisioning.shared_secret"] == "generate":
             base["appservice.provisioning.shared_secret"] = self._new_token()
 
-        copy("appservice.community_id")
-
         copy("metrics.enabled")
         copy("metrics.listen_port")
 

mautrix_twitter/db/message.py

@@ -1,5 +1,5 @@
 # mautrix-twitter - A Matrix-Twitter DM puppeting bridge
-# Copyright (C) 2020 Tulir Asokan
+# Copyright (C) 2022 Tulir Asokan
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -13,14 +13,16 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import TYPE_CHECKING, ClassVar, Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar
 
 from attr import dataclass
 
 from mautrix.types import EventID, RoomID
 from mautrix.util.async_db import Database
 
-fake_db = Database("") if TYPE_CHECKING else None
+fake_db = Database.create("") if TYPE_CHECKING else None
 
 
 @dataclass
@@ -45,23 +47,17 @@ async def delete_all(cls, room_id: RoomID) -> None:
         await cls.db.execute("DELETE FROM message WHERE mx_room=$1", room_id)
 
     @classmethod
-    async def get_by_mxid(cls, mxid: EventID, mx_room: RoomID) -> Optional["Message"]:
-        row = await cls.db.fetchrow(
-            "SELECT mxid, mx_room, twid, receiver " "FROM message WHERE mxid=$1 AND mx_room=$2",
-            mxid,
-            mx_room,
-        )
+    async def get_by_mxid(cls, mxid: EventID, mx_room: RoomID) -> Message | None:
+        q = "SELECT mxid, mx_room, twid, receiver FROM message WHERE mxid=$1 AND mx_room=$2"
+        row = await cls.db.fetchrow(q, mxid, mx_room)
         if not row:
             return None
         return cls(**row)
 
     @classmethod
-    async def get_by_twid(cls, twid: int, receiver: int = 0) -> Optional["Message"]:
-        row = await cls.db.fetchrow(
-            "SELECT mxid, mx_room, twid, receiver " "FROM message WHERE twid=$1 AND receiver=$2",
-            twid,
-            receiver,
-        )
+    async def get_by_twid(cls, twid: int, receiver: int = 0) -> Message | None:
+        q = "SELECT mxid, mx_room, twid, receiver FROM message WHERE twid=$1 AND receiver=$2"
+        row = await cls.db.fetchrow(q, twid, receiver)
         if not row:
             return None
         return cls(**row)

mautrix_twitter/db/portal.py

@@ -1,5 +1,5 @@
 # mautrix-twitter - A Matrix-Twitter DM puppeting bridge
-# Copyright (C) 2020 Tulir Asokan
+# Copyright (C) 2022 Tulir Asokan
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -13,7 +13,9 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import TYPE_CHECKING, ClassVar, List, Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar
 
 from attr import dataclass
 import asyncpg
@@ -22,7 +24,7 @@
 from mautrix.util.async_db import Database
 from mautwitdm.types import ConversationType
 
-fake_db = Database("") if TYPE_CHECKING else None
+fake_db = Database.create("") if TYPE_CHECKING else None
 
 
 @dataclass
@@ -32,18 +34,14 @@ class Portal:
     twid: str
     receiver: int
     conv_type: ConversationType
-    other_user: Optional[int]
-    mxid: Optional[RoomID]
-    name: Optional[str]
+    other_user: int | None
+    mxid: RoomID | None
+    name: str | None
     encrypted: bool
 
-    async def insert(self) -> None:
-        q = (
-            "INSERT INTO portal (twid, receiver, conv_type, other_user, mxid, name, encrypted) "
-            "VALUES ($1, $2, $3, $4, $5, $6, $7)"
-        )
-        await self.db.execute(
-            q,
+    @property
+    def _values(self):
+        return (
             self.twid,
             self.receiver,
             self.conv_type.value,
@@ -53,29 +51,27 @@ async def insert(self) -> None:
             self.encrypted,
         )
 
+    async def insert(self) -> None:
+        q = (
+            "INSERT INTO portal (twid, receiver, conv_type, other_user, mxid, name, encrypted) "
+            "VALUES ($1, $2, $3, $4, $5, $6, $7)"
+        )
+        await self.db.execute(q, *self._values)
+
     async def update(self) -> None:
         q = (
             "UPDATE portal SET conv_type=$3, other_user=$4, mxid=$5, name=$6, encrypted=$7 "
             "WHERE twid=$1 AND receiver=$2"
         )
-        await self.db.execute(
-            q,
-            self.twid,
-            self.receiver,
-            self.conv_type.value,
-            self.other_user,
-            self.mxid,
-            self.name,
-            self.encrypted,
-        )
+        await self.db.execute(q, *self._values)
 
     @classmethod
     def _from_row(cls, row: asyncpg.Record) -> "Portal":
         data = {**row}
         return cls(conv_type=ConversationType(data.pop("conv_type")), **data)
 
     @classmethod
-    async def get_by_mxid(cls, mxid: RoomID) -> Optional["Portal"]:
+    async def get_by_mxid(cls, mxid: RoomID) -> Portal | None:
         q = (
             "SELECT twid, receiver, conv_type, other_user, mxid, name, encrypted "
             "FROM portal WHERE mxid=$1"
@@ -86,7 +82,7 @@ async def get_by_mxid(cls, mxid: RoomID) -> Optional["Portal"]:
         return cls._from_row(row)
 
     @classmethod
-    async def get_by_twid(cls, twid: str, receiver: int = 0) -> Optional["Portal"]:
+    async def get_by_twid(cls, twid: str, receiver: int = 0) -> Portal | None:
         q = (
             "SELECT twid, receiver, conv_type, other_user, mxid, name, encrypted "
             "FROM portal WHERE twid=$1 AND receiver=$2"
@@ -97,7 +93,7 @@ async def get_by_twid(cls, twid: str, receiver: int = 0) -> Optional["Portal"]:
         return cls._from_row(row)
 
     @classmethod
-    async def find_private_chats_of(cls, receiver: int) -> List["Portal"]:
+    async def find_private_chats_of(cls, receiver: int) -> list[Portal]:
         q = (
             "SELECT twid, receiver, conv_type, other_user, mxid, name, encrypted FROM portal "
             "WHERE receiver=$1 AND conv_type='ONE_TO_ONE'"
@@ -106,7 +102,7 @@ async def find_private_chats_of(cls, receiver: int) -> List["Portal"]:
         return [cls._from_row(row) for row in rows]
 
     @classmethod
-    async def find_private_chats_with(cls, other_user: int) -> List["Portal"]:
+    async def find_private_chats_with(cls, other_user: int) -> list[Portal]:
         q = (
             "SELECT twid, receiver, conv_type, other_user, mxid, name, encrypted FROM portal "
             "WHERE other_user=$1 AND conv_type='ONE_TO_ONE'"
@@ -115,7 +111,7 @@ async def find_private_chats_with(cls, other_user: int) -> List["Portal"]:
         return [cls._from_row(row) for row in rows]
 
     @classmethod
-    async def all_with_room(cls) -> List["Portal"]:
+    async def all_with_room(cls) -> list[Portal]:
         q = (
             "SELECT twid, receiver, conv_type, other_user, mxid, name, encrypted FROM portal "
             "WHERE mxid IS NOT NULL"