tobast/matrix-alertbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add typing to every method definition

Browse source
This commit is contained in:
Andrew Morgan 2021-01-09 22:30:07 -05:00
parent 4aa1e2d0f4
commit 2b03c03891
8 changed files with 124 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
my_project_name/bot_commands.py
View file

@ -1,22 +1,34 @@
from nio import AsyncClient, MatrixRoom, RoomMessageText
from my_project_name.chat_functions import react_to_event, send_text_to_room
from my_project_name.config import Config
from my_project_name.storage import Storage
class Command(object):
def __init__(self, client, store, config, command, room, event):
class Command:
def __init__(
self,
client: AsyncClient,
store: Storage,
config: Config,
command: str,
room: MatrixRoom,
event: RoomMessageText,
):
"""A command made by a user
Args:
client (nio.AsyncClient): The client to communicate to matrix with
client: The client to communicate to matrix with
store (Storage): Bot storage
store: Bot storage
config (Config): Bot configuration parameters
config: Bot configuration parameters
command (str): The command and arguments
command: The command and arguments
room (nio.rooms.MatrixRoom): The room the command was sent in
room: The room the command was sent in
event (nio.events.room_events.RoomMessageText): The event describing the command
event: The event describing the command
"""
self.client = client
self.store = store

60
my_project_name/callbacks.py
View file

@ -1,37 +1,47 @@
import logging
from nio import JoinError, MatrixRoom, MegolmEvent, RoomGetEventError, UnknownEvent
from nio import (
AsyncClient,
InviteMemberEvent,
JoinError,
MatrixRoom,
MegolmEvent,
RoomGetEventError,
RoomMessageText,
UnknownEvent,
)
from my_project_name.bot_commands import Command
from my_project_name.chat_functions import make_pill, react_to_event, send_text_to_room
from my_project_name.config import Config
from my_project_name.message_responses import Message
from my_project_name.storage import Storage
logger = logging.getLogger(__name__)
class Callbacks(object):
def __init__(self, client, store, config):
class Callbacks:
def __init__(self, client: AsyncClient, store: Storage, config: Config):
"""
Args:
client (nio.AsyncClient): nio client used to interact with matrix
client: nio client used to interact with matrix
store (Storage): Bot storage
store: Bot storage
config (Config): Bot configuration parameters
config: Bot configuration parameters
"""
self.client = client
self.store = store
self.config = config
self.command_prefix = config.command_prefix
async def message(self, room, event):
async def message(self, room: MatrixRoom, event: RoomMessageText) -> None:
"""Callback for when a message event is received
Args:
room (nio.rooms.MatrixRoom): The room the event came from
event (nio.events.room_events.RoomMessageText): The event defining the message
room: The room the event came from
event: The event defining the message
"""
# Extract the message text
msg = event.body
@ -67,8 +77,14 @@ class Callbacks(object):
command = Command(self.client, self.store, self.config, msg, room, event)
await command.process()
async def invite(self, room, event):
"""Callback for when an invite is received. Join the room specified in the invite"""
async def invite(self, room: MatrixRoom, event: InviteMemberEvent) -> None:
"""Callback for when an invite is received. Join the room specified in the invite.
Args:
room: The room that we are invited to.
event: The invite event.
"""
logger.debug(f"Got invite to {room.room_id} from {event.sender}.")
# Attempt to join 3 times before giving up
@ -90,7 +106,7 @@ class Callbacks(object):
async def _reaction(
self, room: MatrixRoom, event: UnknownEvent, reacted_to_id: str
):
) -> None:
"""A reaction was sent to one of our messages. Let's send a reply acknowledging it.
Args:
@ -130,8 +146,14 @@ class Callbacks(object):
reply_to_event_id=reacted_to_id,
)
async def decryption_failure(self, room: MatrixRoom, event: MegolmEvent):
"""Callback for when an event fails to decrypt. Inform the user"""
async def decryption_failure(self, room: MatrixRoom, event: MegolmEvent) -> None:
"""Callback for when an event fails to decrypt. Inform the user.
Args:
room: The room that the event that we were unable to decrypt is in.
event: The encrypted event that we were unable to decrypt.
"""
logger.error(
f"Failed to decrypt event '{event.event_id}' in room '{room.room_id}'!"
f"\n\n"
@ -152,14 +174,15 @@ class Callbacks(object):
red_x_and_lock_emoji,
)
async def unknown(self, room: MatrixRoom, event: UnknownEvent):
async def unknown(self, room: MatrixRoom, event: UnknownEvent) -> None:
"""Callback for when an event with a type that is unknown to matrix-nio is received.
Currently this is used for reaction events, which are not specced.
Currently this is used for reaction events, which are not yet part of a released
matrix spec (and are thus unknown to nio).
Args:
room: The room the reaction was sent in.
event: The reaction event.
event: The event itself.
"""
if event.type == "m.reaction":
# Get the ID of the event this was a reaction to
@ -168,6 +191,7 @@ class Callbacks(object):
reacted_to = relation_dict.get("event_id")
if reacted_to and relation_dict.get("rel_type") == "m.annotation":
await self._reaction(room, event, reacted_to)
return
logger.debug(
f"Got unknown event with type to {event.type} from {event.sender} in {room.room_id}."

10
my_project_name/chat_functions.py
View file

@ -8,6 +8,7 @@ from nio import (
MatrixRoom,
MegolmEvent,
Response,
RoomSendResponse,
SendRetryError,
)
@ -21,7 +22,7 @@ async def send_text_to_room(
notice: bool = True,
markdown_convert: bool = True,
reply_to_event_id: Optional[str] = None,
):
) -> Union[RoomSendResponse, ErrorResponse]:
"""Send text to a matrix room.
Args:
@ -39,6 +40,9 @@ async def send_text_to_room(
reply_to_event_id: Whether this message is a reply to another event. The event
ID this is message is a reply to.
Returns:
A RoomSendResponse if the request was successful, else an ErrorResponse.
"""
# Determine whether to ping room members or not
msgtype = "m.notice" if notice else "m.text"
@ -56,7 +60,7 @@ async def send_text_to_room(
content["m.relates_to"] = {"m.in_reply_to": {"event_id": reply_to_event_id}}
try:
await client.room_send(
return await client.room_send(
room_id,
"m.room.message",
content,
@ -125,7 +129,7 @@ async def react_to_event(
)
async def decryption_failure(self, room: MatrixRoom, event: MegolmEvent):
async def decryption_failure(self, room: MatrixRoom, event: MegolmEvent) -> None:
"""Callback for when an event fails to decrypt. Inform the user"""
logger.error(
f"Failed to decrypt event '{event.event_id}' in room '{room.room_id}'!"

18
my_project_name/config.py
View file

@ -2,7 +2,7 @@ import logging
import os
import re
import sys
from typing import Any, List
from typing import Any, List, Optional
import yaml
@ -14,11 +14,11 @@ logging.getLogger("peewee").setLevel(
) # Prevent debug messages from peewee lib
class Config(object):
def __init__(self, filepath):
class Config:
def __init__(self, filepath: str):
"""
Args:
filepath (str): Path to config file
filepath: Path to a config file
"""
if not os.path.isfile(filepath):
raise ConfigError(f"Config file '{filepath}' does not exist")
@ -104,15 +104,15 @@ class Config(object):
def _get_cfg(
self,
path: List[str],
default: Any = None,
required: bool = True,
default: Optional[Any] = None,
required: Optional[bool] = True,
) -> Any:
"""Get a config option from a path and option name, specifying whether it is
required.
Raises:
ConfigError: If required is specified and the object is not found
(and there is no default value provided), this error will be raised
ConfigError: If required is True and the object is not found (and there is
no default value provided), a ConfigError will be raised.
"""
# Sift through the the config until we reach our option
config = self.config
@ -128,5 +128,5 @@ class Config(object):
# or return the default value
return default
# We found the option. Return it
# We found the option. Return it.
return config

7
my_project_name/errors.py
View file

@ -1,9 +1,12 @@
# This file holds custom error types that you can define for your application.
class ConfigError(RuntimeError):
"""An error encountered during reading the config file
Args:
msg (str): The message displayed to the user on error
msg: The message displayed to the user on error
"""
def __init__(self, msg):
def __init__(self, msg: str):
super(ConfigError, self).__init__("%s" % (msg,))

1
my_project_name/main.py
View file

@ -112,4 +112,5 @@ async def main():
await client.close()
# Run the main function in an asyncio event loop
asyncio.get_event_loop().run_until_complete(main())

32
my_project_name/message_responses.py
View file

@ -1,26 +1,38 @@
import logging
from nio import AsyncClient, MatrixRoom, RoomMessageText
from my_project_name.chat_functions import send_text_to_room
from my_project_name.config import Config
from my_project_name.storage import Storage
logger = logging.getLogger(__name__)
class Message(object):
def __init__(self, client, store, config, message_content, room, event):
class Message:
def __init__(
self,
client: AsyncClient,
store: Storage,
config: Config,
message_content: str,
room: MatrixRoom,
event: RoomMessageText,
):
"""Initialize a new Message
Args:
client (nio.AsyncClient): nio client used to interact with matrix
client: nio client used to interact with matrix
store (Storage): Bot storage
store: Bot storage
config (Config): Bot configuration parameters
config: Bot configuration parameters
message_content (str): The body of the message
message_content: The body of the message
room (nio.rooms.MatrixRoom): The room the event came from
room: The room the event came from
event (nio.events.room_events.RoomMessageText): The event defining the message
event: The event defining the message
"""
self.client = client
self.store = store
@ -29,12 +41,12 @@ class Message(object):
self.room = room
self.event = event
async def process(self):
async def process(self) -> None:
"""Process and possibly respond to the message"""
if self.message_content.lower() == "hello world":
await self._hello_world()
async def _hello_world(self):
async def _hello_world(self) -> None:
"""Say hello"""
text = "Hello, world!"
await send_text_to_room(self.client, self.room.room_id, text)

26
my_project_name/storage.py
View file

@ -1,4 +1,5 @@
import logging
from typing import Any, Dict
# The latest migration version of the database.
#
@ -12,8 +13,8 @@ latest_migration_version = 0
logger = logging.getLogger(__name__)
class Storage(object):
def __init__(self, database_config):
class Storage:
def __init__(self, database_config: Dict[str, str]):
"""Setup the database
Runs an initial setup or migrations depending on whether a database file has already
@ -45,7 +46,10 @@ class Storage(object):
logger.info(f"Database initialization of type '{self.db_type}' complete")
def _get_database_connection(self, database_type: str, connection_string: str):
def _get_database_connection(
self, database_type: str, connection_string: str
) -> Any:
"""Creates and returns a connection to the database"""
if database_type == "sqlite":
import sqlite3
@ -61,7 +65,7 @@ class Storage(object):
return conn
def _initial_setup(self):
def _initial_setup(self) -> None:
"""Initial setup of the database"""
logger.info("Performing initial database setup...")
@ -88,13 +92,13 @@ class Storage(object):
logger.info("Database setup complete")
def _run_migrations(self, current_migration_version: int):
def _run_migrations(self, current_migration_version: int) -> None:
"""Execute database migrations. Migrates the database to the
`latest_migration_version`
Args:
current_migration_version: The migration version that the database is
currently at
currently at.
"""
logger.debug("Checking for necessary database migrations...")
@ -108,8 +112,14 @@ class Storage(object):
#
# logger.info("Database migrated to v1")
def _execute(self, *args):
"""A wrapper around cursor.execute that transforms placeholder ?'s to %s for postgres"""
def _execute(self, *args) -> None:
"""A wrapper around cursor.execute that transforms placeholder ?'s to %s for postgres.
This allows for the support of queries that are compatible with both postgres and sqlite.
Args:
args: Arguments passed to cursor.execute.
"""
if self.db_type == "postgres":
self.cursor.execute(args[0].replace("?", "%s"), *args[1:])
else: