Firepup650/bitbot-3.11-fork
2
0
Fork 0
Code Issues Pull requests Activity Actions

Fix type errors detected by 'mypy --ignore-missing-imports src'.

Browse source
This commit is contained in:
Valentin Lorentz 2019-10-26 00:32:00 +02:00
parent 9958c9169d
commit fbf8cd1a16
10 changed files with 51 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/Control.py
View file

@ -12,7 +12,7 @@ class ControlClient(object):
def fileno(self) -> int:
return self._socket.fileno()
def read_lines(self) -> typing.List[str]:
def read_lines(self) -> typing.Optional[typing.List[str]]:
try:
data = self._socket.recv(2048)
except:
@ -45,11 +45,11 @@ class Control(PollSource.PollSource):
self._filename = filename
self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self._clients = {}
self._clients: typing.Dict[int, ControlClient] = {}
def _on_log(self, levelno: int, line: str):
for client in self._clients.values():
if not client.log_level == None and client.log_level <= levelno:
if client.log_level is not None and client.log_level <= levelno:
self._send_action(client, "log", line)
def bind(self):
@ -69,7 +69,7 @@ class Control(PollSource.PollSource):
elif fileno in self._clients:
client = self._clients[fileno]
lines = client.read_lines()
if lines == None:
if lines is None:
client.disconnect()
del self._clients[fileno]
else:
@ -109,7 +109,7 @@ class Control(PollSource.PollSource):
if not keepalive:
client.disconnect()
def _send_action(self, client: ControlClient, action: str, data: str,
id: int=None):
def _send_action(self, client: ControlClient, action: str,
data: typing.Optional[str], id: typing.Optional[str]=None):
client.write_line(
json.dumps({"action": action, "data": data, "id": id}))

4
src/EventManager.py
View file

@ -37,8 +37,8 @@ class EventHook(object):
self.priority = priority
self.docstring = utils.parse.docstring(func.__doc__ or "")
self._kwargs = {}
self._multi_kwargs = {}
self._kwargs: typing.Dict[str, typing.Any] = {}
self._multi_kwargs: typing.Dict[str, typing.List[typing.Any]] = {}
for key, value in kwargs:
if key in self._multi_kwargs:
self._multi_kwargs[key].append(value)

6
src/IRCBuffer.py
View file

@ -67,7 +67,11 @@ class Buffer(object):
return None
def find_from(self, nickname: str) -> typing.Optional[BufferLine]:
return (self.find_many_from(nickname, 1) or [None])[0]
lines = self.find_many_from(nickname, 1)
if lines:
return lines[0]
else:
return None
def find_many_from(self, nickname: str, max: int
) -> typing.List[BufferLine]:
nickname_lower = self.server.irc_lower(nickname)

4
src/IRCLine.py
View file

@ -1,5 +1,5 @@
import datetime, typing, uuid
from src import IRCObject, utils
from src import EventManager, IRCObject, utils
# this should be 510 (RFC1459, 512 with \r\n) but a server BitBot uses is broken
LINE_MAX = 470
@ -130,7 +130,7 @@ class ParsedLine(object):
def _newline_truncate(self, line: str) -> typing.Tuple[str, str]:
line, sep, overflow = line.partition("\n")
return [line, overflow]
return (line, overflow)
def _line_max(self, hostmask: str, margin: int) -> int:
return LINE_MAX-len((":%s " % hostmask).encode("utf8"))-margin
def truncate(self, hostmask: str, margin: int=0) -> typing.Tuple[str, str]:

6
src/IRCServer.py
View file

@ -166,7 +166,7 @@ class Server(IRCObject.Object):
def has_user(self, nickname: str) -> bool:
return self.irc_lower(nickname) in self.users
def get_user(self, nickname: str, username: str=None, hostname: str=None,
def get_user(self, nickname: str, username: typing.Optional[str]=None, hostname: str=None,
create: bool=True) -> typing.Optional[IRCUser.User]:
new = False
if not self.has_user(nickname) and create:
@ -178,9 +178,9 @@ class Server(IRCObject.Object):
user = self.users.get(self.irc_lower(nickname), None)
if user:
if not username == None:
if username is not None:
user.username = username
if not hostname == None:
if hostname is not None:
user.hostname = hostname
if new:
self.events.on("new.user").call(user=new_user, server=self)

12
src/IRCUser.py
View file

@ -11,11 +11,11 @@ class User(IRCObject.Object):
self.server = server
self.set_nickname(nickname)
self._id = id
self.username = None
self.hostname = None
self.realname = None
self.username: typing.Optional[str] = None
self.hostname: typing.Optional[str] = None
self.realname: typing.Optional[str] = None
self.bot = bot
self.channels = set([]) # type: typing.Set[IRCChannel.Channel]
self.channels: typing.Set[IRCChannel.Channel] = set([])
self.identified_account = None
self.identified_account_override = None
@ -23,7 +23,7 @@ class User(IRCObject.Object):
self.identified_account_id = None
self.identified_account_id_override = None
self.away = False
self.away_message = None # type: typing.Optional[str]
self.away_message: typing.Optional[str] = None
self.buffer = IRCBuffer.Buffer(bot, server)
@ -32,7 +32,7 @@ class User(IRCObject.Object):
def __str__(self) -> str:
return self.nickname
def hostmask(self) -> str:
def hostmask(self) -> typing.Optional[str]:
if self.nickname and self.username and self.hostname:
return "%s!%s@%s" % (self.nickname, self.username, self.hostname)
return None

2
src/Logging.py
View file

@ -24,6 +24,8 @@ class HookedHandler(logging.StreamHandler):
self._func(record.levelno, self.format(record))
class Log(object):
_hooks: typing.List[typing.Callable[[int, str], None]]
def __init__(self, to_file: bool, level: str, location: str):
self._hooks = []

11
src/ModuleManager.py
View file

@ -1,4 +1,5 @@
import enum, gc, glob, importlib, io, inspect, os, sys, typing, uuid
import enum, gc, glob, importlib, importlib.util, io, inspect, os, sys
import typing, uuid
from src import Config, EventManager, Exports, IRCBot, Logging, Timers, utils
class ModuleException(Exception):
@ -155,7 +156,7 @@ class ModuleManager(object):
def _module_name(self, path: str) -> str:
return os.path.basename(path).rsplit(".py", 1)[0].lower()
def _module_paths(self, name: str) -> str:
def _module_paths(self, name: str) -> typing.List[str]:
paths = []
for directory in self.directories:
paths.append(os.path.join(directory, name))
@ -180,7 +181,7 @@ class ModuleManager(object):
return getattr(obj, magic) if hasattr(obj, magic) else default
def _check_hashflags(self, bot: "IRCBot.Bot", definition: ModuleDefinition
) -> bool:
) -> None:
for hashflag, value in definition.hashflags:
if hashflag == "ignore":
# nope, ignore this module.
@ -321,7 +322,7 @@ class ModuleManager(object):
def load_modules(self, bot: "IRCBot.Bot", whitelist: typing.List[str]=[],
blacklist: typing.List[str]=[]
) -> typing.Tuple[typing.List[str], typing.List[str]]:
) -> None:
loadable, nonloadable = self._list_valid_modules(bot, whitelist, blacklist)
for definition in nonloadable:
@ -388,7 +389,7 @@ class ModuleManager(object):
failed = (definition, e)
break
if not failed == None:
if failed is not None:
for module in self.modules.values():
self._unload_module(module)
self.modules = old_modules

26
src/utils/__init__.py
View file

@ -177,7 +177,7 @@ class BitBotMagic(object):
self._kwargs.append((key, value))
def get_hooks(self):
hooks: typing.List[typing.Tuple[str, typing.List[str, typing.Any]]] = []
hooks: typing.List[typing.Tuple[str, typing.List[Tuple[str, typing.Any]]]] = []
for hook, kwargs in self._hooks:
hooks.append((hook, self._kwargs.copy()+list(kwargs.items())))
return hooks
@ -254,8 +254,11 @@ class CaseInsensitiveDict(dict):
return dict.__getitem__(self, key.lower())
def __setitem__(self, key: str, value: typing.Any) -> typing.Any:
return dict.__setitem__(self, key.lower(), value)
def __contains__(self, key: str):
def __contains__(self, key: typing.Any) -> bool:
if isinstance(key, str):
return dict.__contains__(self, key.lower())
else:
raise TypeError('Expected string, not %r' % (key,))
def get(self, key: str, default: typing.Any=None):
return dict.get(self, key.lower(), default)
@ -296,7 +299,7 @@ class Setting(object):
SETTING_TRUE = ["true", "yes", "on", "y"]
SETTING_FALSE = ["false", "no", "off", "n"]
class BoolSetting(Setting):
example = "on"
example: typing.Optional[str] = "on"
def parse(self, value: str) -> typing.Any:
value_lower = value.lower()
if value_lower in SETTING_TRUE:
@ -306,7 +309,7 @@ class BoolSetting(Setting):
return None
class IntSetting(Setting):
example = "10"
example: typing.Optional[str] = "10"
def parse(self, value: str) -> typing.Any:
if value == "0":
return 0
@ -317,7 +320,7 @@ class IntSetting(Setting):
return None
class IntRangeSetting(IntSetting):
example = None
example: typing.Optional[str] = None
def __init__(self, n_min: int, n_max: int, name: str, help: str=None,
example: str=None):
self._n_min = n_min
@ -365,7 +368,7 @@ class FunctionSetting(Setting):
self._func = func
Setting.__init__(self, name, help, example)
if not format == None:
self.format = format
self.format = format # type: ignore
def parse(self, value: str) -> typing.Any:
return self._func(value)
@ -399,8 +402,11 @@ def deadline(seconds: int=10):
signal.signal(signal.SIGALRM, old_handler)
signal.setitimer(signal.ITIMER_REAL, old_seconds, 0)
def deadline_process(func: typing.Callable[[], None], seconds: int=10):
q = multiprocessing.Queue()
DeadlineProcessReturnType = typing.TypeVar('DeadlineProcessReturnType')
def deadline_process(func: typing.Callable[[], DeadlineProcessReturnType],
seconds: int=10) -> DeadlineProcessReturnType:
q: multiprocessing.Queue[typing.Tuple[bool, DeadlineProcessReturnType]] = \
multiprocessing.Queue()
def _wrap(func, q):
try:
q.put([True, func()])
@ -414,7 +420,7 @@ def deadline_process(func: typing.Callable[[], None], seconds: int=10):
try:
success, out = q.get(block=True, timeout=seconds)
except queue.Empty:
p.kill()
p.kill() # type: ignore # to make mypy pass on Python 3.6
deadlined = True
finally:
q.close()
@ -425,4 +431,4 @@ def deadline_process(func: typing.Callable[[], None], seconds: int=10):
if success:
return out
else:
raise out
raise out # type: ignore

4
src/utils/http.py
View file

@ -162,13 +162,13 @@ def _find_encoding(soup: bs4.BeautifulSoup) -> typing.Optional[str]:
return None
def request(request_obj: typing.Union[str, Request], **kwargs) -> Response:
if type(request_obj) == str:
if isinstance(request_obj, str):
request_obj = Request(request_obj, **kwargs)
return _request(request_obj)
def _request(request_obj: Request) -> Response:
def _wrap():
def _wrap() -> Response:
headers = request_obj.get_headers()
response = requests.request(
request_obj.method,