logger: Add typing information.

This commit is contained in:
Emmanuel Gil Peyrot 2018-07-22 16:38:32 +02:00
parent e9585c7ea1
commit 07b534dc39

View file

@ -11,6 +11,7 @@ conversations and roster changes
import mmap import mmap
import re import re
from typing import List, Dict, Optional, TextIO, BinaryIO, Any
from datetime import datetime from datetime import datetime
from poezio import common from poezio import common
@ -73,9 +74,9 @@ class Logger:
""" """
def __init__(self): def __init__(self):
self._roster_logfile = None self._roster_logfile = None # Optional[TextIO]
# a dict of 'groupchatname': file-object (opened) # a dict of 'groupchatname': file-object (opened)
self._fds = {} self._fds = {} # type: Dict[str, TextIO]
def __del__(self): def __del__(self):
for opened_file in self._fds.values(): for opened_file in self._fds.values():
@ -85,14 +86,14 @@ class Logger:
except: # Can't close? too bad except: # Can't close? too bad
pass pass
def close(self, jid): def close(self, jid) -> None:
jid = str(jid).replace('/', '\\') jid = str(jid).replace('/', '\\')
if jid in self._fds: if jid in self._fds:
self._fds[jid].close() self._fds[jid].close()
log.debug('Log file for %s closed.', jid) log.debug('Log file for %s closed.', jid)
del self._fds[jid] del self._fds[jid]
def reload_all(self): def reload_all(self) -> None:
"""Close and reload all the file handles (on SIGHUP)""" """Close and reload all the file handles (on SIGHUP)"""
for opened_file in self._fds.values(): for opened_file in self._fds.values():
if opened_file: if opened_file:
@ -102,7 +103,7 @@ class Logger:
self._fds[room] = self._check_and_create_log_dir(room) self._fds[room] = self._check_and_create_log_dir(room)
log.debug('Log handle for %s re-created', room) log.debug('Log handle for %s re-created', room)
def _check_and_create_log_dir(self, room, open_fd=True): def _check_and_create_log_dir(self, room: str, open_fd: bool = True) -> Optional[TextIO]:
""" """
Check that the directory where we want to log the messages Check that the directory where we want to log the messages
exists. if not, create it exists. if not, create it
@ -127,7 +128,7 @@ class Logger:
log.error( log.error(
'Unable to open the log file (%s)', filename, exc_info=True) 'Unable to open the log file (%s)', filename, exc_info=True)
def get_logs(self, jid, nb=10): def get_logs(self, jid: str, nb: int = 10) -> Optional[List[Dict[str, Any]]]:
""" """
Get the nb last messages from the log history for the given jid. Get the nb last messages from the log history for the given jid.
Note that a message may be more than one line in these files, so Note that a message may be more than one line in these files, so
@ -172,7 +173,7 @@ class Logger:
return return
return parse_log_lines(lines) return parse_log_lines(lines)
def log_message(self, jid, nick, msg, date=None, typ=1): def log_message(self, jid: str, nick: str, msg: str, date: Optional[datetime] = None, typ: int = 1) -> bool:
""" """
log the message in the appropriate jid's file log the message in the appropriate jid's file
type: type:
@ -211,7 +212,7 @@ class Logger:
return False return False
return True return True
def log_roster_change(self, jid, message): def log_roster_change(self, jid: str, message: str) -> bool:
""" """
Log a roster change Log a roster change
""" """
@ -248,7 +249,7 @@ class Logger:
return True return True
def build_log_message(nick, msg, date=None, typ=1): def build_log_message(nick: str, msg: str, date: Optional[datetime] = None, typ: int = 1) -> str:
""" """
Create a log message from a nick, a message, optionally a date and type Create a log message from a nick, a message, optionally a date and type
message types: message types:
@ -275,7 +276,7 @@ def build_log_message(nick, msg, date=None, typ=1):
return logged_msg + ''.join(' %s\n' % line for line in lines) return logged_msg + ''.join(' %s\n' % line for line in lines)
def get_lines_from_fd(fd, nb=10): def get_lines_from_fd(fd: BinaryIO, nb: int = 10) -> List[str]:
""" """
Get the last log lines from a fileno Get the last log lines from a fileno
""" """
@ -294,7 +295,7 @@ def get_lines_from_fd(fd, nb=10):
return lines return lines
def parse_log_lines(lines): def parse_log_lines(lines: List[str]) -> List[Dict[str, Any]]:
""" """
Parse raw log lines into poezio log objects Parse raw log lines into poezio log objects
""" """
@ -334,7 +335,7 @@ def parse_log_lines(lines):
return messages return messages
def create_logger(): def create_logger() -> None:
"Create the global logger object" "Create the global logger object"
global logger global logger
logger = Logger() logger = Logger()