pep/poezio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

text_buffer: Add typing information.

Browse source
This commit is contained in:
Emmanuel Gil Peyrot 2018-07-22 18:24:18 +02:00
parent 33718e56c4
commit f107ecd953
1 changed files with 49 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

96
poezio/text_buffer.py
View file

@ -11,9 +11,11 @@ independently by their TextWins.
import logging import logging
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
from typing import Union, Optional, List, Tuple
from datetime import datetime from datetime import datetime
from poezio.config import config from poezio.config import config
from poezio.theming import get_theme, dump_tuple from poezio.theming import get_theme, dump_tuple
from poezio.windows import BaseTextWin
class Message: class Message:
@ -22,24 +24,24 @@ class Message:
'jid', 'ack') 'jid', 'ack')
def __init__(self, def __init__(self,
txt, txt: str,
time, time: Optional[datetime],
nickname, nickname: Optional[str],
nick_color, nick_color: Optional[Tuple],
history, history: bool,
user, user: Optional[str],
identifier, identifier: Optional[str],
str_time=None, str_time: Optional[str] = None,
highlight=False, highlight: bool = False,
old_message=None, old_message: Optional[Message] = None,
revisions=0, revisions: int = 0,
jid=None, jid: Optional[str] = None,
ack=0): ack: int = 0) -> None:
""" """
Create a new Message object with parameters, check for /me messages, Create a new Message object with parameters, check for /me messages,
and delayed messages and delayed messages
""" """
time = time or datetime.now() time = time if time is not None else datetime.now()
if txt.startswith('/me '): if txt.startswith('/me '):
me = True me = True
txt = '\x19%s}%s\x19o' % (dump_tuple(get_theme().COLOR_ME_MESSAGE), txt = '\x19%s}%s\x19o' % (dump_tuple(get_theme().COLOR_ME_MESSAGE),
@ -71,7 +73,7 @@ class Message:
self.jid = jid self.jid = jid
self.ack = ack self.ack = ack
def _other_elems(self): def _other_elems(self) -> str:
"Helper for the repr_message function" "Helper for the repr_message function"
acc = [] acc = []
fields = list(self.__slots__) fields = list(self.__slots__)
@ -80,7 +82,7 @@ class Message:
acc.append('%s=%s' % (field, repr(getattr(self, field)))) acc.append('%s=%s' % (field, repr(getattr(self, field))))
return 'Message(%s, %s' % (', '.join(acc), 'old_message=') return 'Message(%s, %s' % (', '.join(acc), 'old_message=')
def __repr__(self): def __repr__(self) -> str:
""" """
repr() for the Message class, for debug purposes, since the default repr() for the Message class, for debug purposes, since the default
repr() is recursive, so it can stack overflow given too many revisions repr() is recursive, so it can stack overflow given too many revisions
@ -115,37 +117,37 @@ class TextBuffer:
information and attributes. information and attributes.
""" """
def __init__(self, messages_nb_limit=None): def __init__(self, messages_nb_limit: Optional[int] = None) -> None:
if messages_nb_limit is None: if messages_nb_limit is None:
messages_nb_limit = config.get('max_messages_in_memory') messages_nb_limit = config.get('max_messages_in_memory')
self._messages_nb_limit = messages_nb_limit self._messages_nb_limit = messages_nb_limit # type: int
# Message objects # Message objects
self.messages = [] self.messages = [] # type: List[Message]
# we keep track of one or more windows # we keep track of one or more windows
# so we can pass the new messages to them, as they are added, so # so we can pass the new messages to them, as they are added, so
# they (the windows) can build the lines from the new message # they (the windows) can build the lines from the new message
self._windows = [] self._windows = [] # type: List[BaseTextWin]
def add_window(self, win): def add_window(self, win: BaseTextWin) -> None:
self._windows.append(win) self._windows.append(win)
@property @property
def last_message(self): def last_message(self) -> Optional[Message]:
return self.messages[-1] if self.messages else None return self.messages[-1] if self.messages else None
def add_message(self, def add_message(self,
txt, txt: str,
time=None, time: Optional[datetime] = None,
nickname=None, nickname: Optional[str] = None,
nick_color=None, nick_color: Optional[Tuple] = None,
history=None, history: bool = False,
user=None, user: Optional[str] = None,
highlight=False, highlight: bool = False,
identifier=None, identifier: Optional[str] = None,
str_time=None, str_time: Optional[str] = None,
jid=None, jid: Optional[str] = None,
ack=0): ack: int = 0) -> int:
""" """
Create a message and add it to the text buffer Create a message and add it to the text buffer
""" """
@ -184,7 +186,7 @@ class TextBuffer:
return min(ret_val, 1) return min(ret_val, 1)
def _find_message(self, old_id): def _find_message(self, old_id: str) -> int:
""" """
Find a message in the text buffer from its message id Find a message in the text buffer from its message id
""" """
@ -194,22 +196,22 @@ class TextBuffer:
return i return i
return -1 return -1
def ack_message(self, old_id, jid): def ack_message(self, old_id: str, jid: str) -> Union[None, bool, Message]:
"""Mark a message as acked""" """Mark a message as acked"""
return self._edit_ack(1, old_id, jid) return self._edit_ack(1, old_id, jid)
def nack_message(self, error, old_id, jid): def nack_message(self, error: str, old_id: str, jid: str) -> Union[None, bool, Message]:
"""Mark a message as errored""" """Mark a message as errored"""
return self._edit_ack(-1, old_id, jid, append=error) return self._edit_ack(-1, old_id, jid, append=error)
def _edit_ack(self, value, old_id, jid, append=''): def _edit_ack(self, value: int, old_id: str, jid: str, append: str = '') -> Union[None, bool, Message]:
""" """
Edit the ack status of a message, and optionally Edit the ack status of a message, and optionally
append some text. append some text.
""" """
i = self._find_message(old_id) i = self._find_message(old_id)
if i == -1: if i == -1:
return return None
msg = self.messages[i] msg = self.messages[i]
if msg.ack == 1: # Message was already acked if msg.ack == 1: # Message was already acked
return False return False
@ -223,13 +225,13 @@ class TextBuffer:
return msg return msg
def modify_message(self, def modify_message(self,
txt, txt: str,
old_id, old_id: str,
new_id, new_id: str,
highlight=False, highlight: bool = False,
time=None, time: Optional[datetime] = None,
user=None, user: Optional[str] = None,
jid=None): jid: Optional[str] = None):
""" """
Correct a message in a text buffer. Correct a message in a text buffer.
""" """
@ -263,7 +265,7 @@ class TextBuffer:
time, time,
msg.nickname, msg.nickname,
msg.nick_color, msg.nick_color,
None, False,
msg.user, msg.user,
new_id, new_id,
highlight=highlight, highlight=highlight,
@ -274,7 +276,7 @@ class TextBuffer:
log.debug('Replacing message %s with %s.', old_id, new_id) log.debug('Replacing message %s with %s.', old_id, new_id)
return message return message
def del_window(self, win): def del_window(self, win: BaseTextWin) -> None:
self._windows.remove(win) self._windows.remove(win)
def __del__(self): def __del__(self):