bots: Add mypy annotations for bots framework.

This commit adds mypy annotations for both the main
bots and the bots testing runner. It involves a change
to the BotHandlerApi send_message and update_message
funtions, which is compatible with every bot.

Tweaked by tabbott to use more expressive annotations.
This commit is contained in:
derAnfaenger 2017-05-26 06:58:19 +02:00 committed by Tim Abbott
parent bd4be69148
commit 8f2f2d220e
4 changed files with 53 additions and 12 deletions

View file

@ -7,6 +7,11 @@ import sys
import time
import re
if False:
from mypy_extensions import NoReturn
from typing import Any, Optional, List, Dict
from types import ModuleType
our_dir = os.path.dirname(os.path.abspath(__file__))
# For dev setups, we can find the API in the repo itself.
@ -16,17 +21,20 @@ if os.path.exists(os.path.join(our_dir, '../api/zulip')):
from zulip import Client
def exit_gracefully(signum, frame):
# type: (int, Optional[Any]) -> None
sys.exit(0)
class RateLimit(object):
def __init__(self, message_limit, interval_limit):
# type: (int, int) -> None
self.message_limit = message_limit
self.interval_limit = interval_limit
self.message_list = []
self.message_list = [] # type: List[float]
self.error_message = '-----> !*!*!*MESSAGE RATE LIMIT REACHED, EXITING*!*!*! <-----\n'
'Is your bot trapped in an infinite loop by reacting to its own messages?'
def is_legal(self):
# type: () -> bool
self.message_list.append(time.time())
if len(self.message_list) > self.message_limit:
self.message_list.pop(0)
@ -36,12 +44,14 @@ class RateLimit(object):
return True
def show_error_and_exit(self):
# type: () -> NoReturn
logging.error(self.error_message)
sys.exit(1)
class BotHandlerApi(object):
def __init__(self, client):
# type: (Client) -> None
# Only expose a subset of our Client's functionality
user_profile = client.get_profile()
self._rate_limit = RateLimit(20, 5)
@ -54,19 +64,22 @@ class BotHandlerApi(object):
' up the zuliprc file correctly.')
sys.exit(1)
def send_message(self, *args, **kwargs):
def send_message(self, message):
# type: (Dict[str, Any]) -> Dict[str, Any]
if self._rate_limit.is_legal():
return self._client.send_message(*args, **kwargs)
return self._client.send_message(message)
else:
self._rate_limit.show_error_and_exit()
def update_message(self, *args, **kwargs):
def update_message(self, message):
# type: (Dict[str, Any]) -> Dict[str, Any]
if self._rate_limit.is_legal():
return self._client.update_message(*args, **kwargs)
return self._client.update_message(message)
else:
self._rate_limit.show_error_and_exit()
def send_reply(self, message, response):
# type: (Dict[str, Any], str) -> Dict[str, Any]
if message['type'] == 'private':
return self.send_message(dict(
type='private',
@ -83,15 +96,25 @@ class BotHandlerApi(object):
class StateHandler(object):
def __init__(self):
self.state = None
# type: () -> None
self.state = None # type: Any
def set_state(self, state):
# type: (Any) -> None
self.state = state
def get_state(self):
# type: () -> Any
return self.state
def run_message_handler_for_bot(lib_module, quiet, config_file):
# type: (Any, bool, str) -> Any
#
# lib_module is of type Any, since it can contain any bot's
# handler class. Eventually, we want bot's handler classes to
# inherit from a common prototype specifying the handle_message
# function.
#
# Make sure you set up your ~/.zuliprc
client = Client(config_file=config_file)
restricted_client = BotHandlerApi(client)
@ -104,6 +127,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
print(message_handler.usage())
def extract_query_without_mention(message, client):
# type: (Dict[str, Any], BotHandlerApi) -> str
"""
If the bot is the first @mention in the message, then this function returns
the message with the bot's @mention removed. Otherwise, it returns None.
@ -116,6 +140,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
return query_without_mention.lstrip()
def is_private(message, client):
# type: (Dict[str, Any], BotHandlerApi) -> bool
# bot will not reply if the sender name is the same as the bot name
# to prevent infinite loop
if message['type'] == 'private':
@ -123,6 +148,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
return False
def handle_message(message):
# type: (Dict[str, Any]) -> None
logging.info('waiting for next message')
# is_mentioned is true if the bot is mentioned at ANY position (not necessarily