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

Add more return type annotations to src/IRCServer.Server

Browse source
This commit is contained in:
jesopo 2018-11-21 20:13:56 +00:00
parent 9a5ba99840
commit bab4fca287
1 changed files with 27 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
src/IRCServer.py
View file

@ -1,5 +1,6 @@
import collections, socket, ssl, sys, time, typing import collections, socket, ssl, sys, time, typing
from src import EventManager, IRCBot, IRCChannels, IRCObject, IRCUser, utils from src import EventManager, IRCBot, IRCChannel, IRCChannels, IRCObject
from src import IRCUser, utils
THROTTLE_LINES = 4 THROTTLE_LINES = 4
THROTTLE_SECONDS = 1 THROTTLE_SECONDS = 1
@ -62,15 +63,15 @@ class Server(IRCObject.Object):
self.events.on("timer.rejoin").hook(self.try_rejoin) self.events.on("timer.rejoin").hook(self.try_rejoin)
def __repr__(self): def __repr__(self) -> str:
return "IRCServer.Server(%s)" % self.__str__() return "IRCServer.Server(%s)" % self.__str__()
def __str__(self): def __str__(self) -> str:
if self.alias: if self.alias:
return self.alias return self.alias
return "%s:%s%s" % (self.connection_params.hostname, return "%s:%s%s" % (self.connection_params.hostname,
"+" if self.connection_params.tls else "", "+" if self.connection_params.tls else "",
self.connection_params.port) self.connection_params.port)
def fileno(self): def fileno(self) -> int:
return self.cached_fileno or self.socket.fileno() return self.cached_fileno or self.socket.fileno()
def tls_wrap(self): def tls_wrap(self):
@ -131,31 +132,35 @@ class Server(IRCObject.Object):
def set_setting(self, setting: str, value: typing.Any): def set_setting(self, setting: str, value: typing.Any):
self.bot.database.server_settings.set(self.id, setting, self.bot.database.server_settings.set(self.id, setting,
value) value)
def get_setting(self, setting: str, default: typing.Any=None): def get_setting(self, setting: str, default: typing.Any=None
) -> typing.Any:
return self.bot.database.server_settings.get(self.id, return self.bot.database.server_settings.get(self.id,
setting, default) setting, default)
def find_settings(self, pattern: str, default: typing.Any=[]): def find_settings(self, pattern: str, default: typing.Any=[]
) -> typing.List[typing.Any]:
return self.bot.database.server_settings.find(self.id, return self.bot.database.server_settings.find(self.id,
pattern, default) pattern, default)
def find_settings_prefix(self, prefix: str, default: typing.Any=[]): def find_settings_prefix(self, prefix: str, default: typing.Any=[]
) -> typing.List[typing.Any]:
return self.bot.database.server_settings.find_prefix( return self.bot.database.server_settings.find_prefix(
self.id, prefix, default) self.id, prefix, default)
def del_setting(self, setting: str): def del_setting(self, setting: str):
self.bot.database.server_settings.delete(self.id, setting) self.bot.database.server_settings.delete(self.id, setting)
def get_user_setting(self, nickname: str, setting: str, def get_user_setting(self, nickname: str, setting: str,
default: typing.Any=None): default: typing.Any=None) -> typing.Any:
user_id = self.get_user_id(nickname) user_id = self.get_user_id(nickname)
return self.bot.database.user_settings.get(user_id, setting, default) return self.bot.database.user_settings.get(user_id, setting, default)
def set_user_setting(self, nickname: str, setting: str, value: typing.Any): def set_user_setting(self, nickname: str, setting: str, value: typing.Any):
user_id = self.get_user_id(nickname) user_id = self.get_user_id(nickname)
self.bot.database.user_settings.set(user_id, setting, value) self.bot.database.user_settings.set(user_id, setting, value)
def get_all_user_settings(self, setting: str, default: typing.Any=[]): def get_all_user_settings(self, setting: str, default: typing.Any=[]
) -> typing.List[typing.Any]:
return self.bot.database.user_settings.find_all_by_setting( return self.bot.database.user_settings.find_all_by_setting(
self.id, setting, default) self.id, setting, default)
def find_all_user_channel_settings(self, setting: str, def find_all_user_channel_settings(self, setting: str,
default: typing.Any=[]): default: typing.Any=[]) -> typing.List[typing.Any]:
return self.bot.database.user_channel_settings.find_all_by_setting( return self.bot.database.user_channel_settings.find_all_by_setting(
self.id, setting, default) self.id, setting, default)
@ -195,7 +200,8 @@ class Server(IRCObject.Object):
channel.remove_user(user) channel.remove_user(user)
def get_target(self, name: str def get_target(self, name: str
) -> typing.Union[IRCChannel.Channel, IRCUser.User]: ) -> typing.Optional[
typing.Union[IRCChannel.Channel, IRCUser.User]]:
if name[0] in self.channel_types: if name[0] in self.channel_types:
if name in self.channels: if name in self.channels:
return self.channels.get(name) return self.channels.get(name)
@ -218,7 +224,7 @@ class Server(IRCObject.Object):
if not len(user.channels): if not len(user.channels):
self.remove_user(user) self.remove_user(user)
self.new_users.clear() self.new_users.clear()
def read(self): def read(self) -> typing.Optional[typing.List[str]]:
data = b"" data = b""
try: try:
data = self.socket.recv(4096) data = self.socket.recv(4096)
@ -259,18 +265,18 @@ class Server(IRCObject.Object):
self.ping_sent = False self.ping_sent = False
return decoded_lines return decoded_lines
def until_next_ping(self): def until_next_ping(self) -> typing.Optional[float]:
if self.ping_sent: if self.ping_sent:
return None return None
return max(0, (self.last_read+PING_INTERVAL_SECONDS return max(0, (self.last_read+PING_INTERVAL_SECONDS
)-time.monotonic()) )-time.monotonic())
def ping_due(self): def ping_due(self) -> bool:
return self.until_next_ping() == 0 return self.until_next_ping() == 0
def until_read_timeout(self): def until_read_timeout(self) -> typing.float:
return max(0, (self.last_read+READ_TIMEOUT_SECONDS return max(0, (self.last_read+READ_TIMEOUT_SECONDS
)-time.monotonic()) )-time.monotonic())
def read_timed_out(self): def read_timed_out(self) -> bool:
return self.until_read_timeout == 0 return self.until_read_timeout == 0
def send(self, data: str): def send(self, data: str):
@ -295,11 +301,11 @@ class Server(IRCObject.Object):
now = time.monotonic() now = time.monotonic()
self.recent_sends.append(now) self.recent_sends.append(now)
self.last_send = now self.last_send = now
def waiting_send(self): def waiting_send(self) -> bool:
return bool(len(self.write_buffer)) or bool(len(self.buffered_lines)) return bool(len(self.write_buffer)) or bool(len(self.buffered_lines))
def throttle_done(self): def throttle_done(self) -> bool:
return self.send_throttle_timeout() == 0 return self.send_throttle_timeout() == 0
def send_throttle_timeout(self): def send_throttle_timeout(self) -> float:
if len(self.write_buffer) or not self._write_throttling: if len(self.write_buffer) or not self._write_throttling:
return 0 return 0
@ -347,7 +353,7 @@ class Server(IRCObject.Object):
def send_starttls(self): def send_starttls(self):
self.send("STARTTLS") self.send("STARTTLS")
def waiting_for_capabilities(self): def waiting_for_capabilities(self) -> bool:
return bool(len(self._capabilities_waiting)) return bool(len(self._capabilities_waiting))
def wait_for_capability(self, capability: str): def wait_for_capability(self, capability: str):
self._capabilities_waiting.add(capability) self._capabilities_waiting.add(capability)
@ -377,7 +383,7 @@ class Server(IRCObject.Object):
def send_quit(self, reason: str="Leaving"): def send_quit(self, reason: str="Leaving"):
self.send("QUIT :%s" % reason) self.send("QUIT :%s" % reason)
def _tag_str(self, tags: dict): def _tag_str(self, tags: dict) -> str:
tag_str = "" tag_str = ""
for tag, value in tags.items(): for tag, value in tags.items():
if tag_str: if tag_str: