Commit 0255ecb5 authored by Emmanuel Gil Peyrot's avatar Emmanuel Gil Peyrot Committed by Philipp Hörist

structs: Migrate from namedtuple to dataclass

This lets us do some typing for the attributes, and specify the default
values in a much more user-friendly way.

This can be done now since we’re on Python 3.7!
parent b1553dde
Pipeline #5420 failed with stages
in 19 seconds
......@@ -17,7 +17,10 @@
import time
import random
from dataclasses import dataclass
from collections import namedtuple
from typing import Dict, Optional, Any, Tuple, List, Callable
from datetime import datetime
from gi.repository import Soup
from gi.repository import Gio
......@@ -40,125 +43,354 @@ from nbxmpp.const import StatusCode
from nbxmpp.const import PresenceType
from nbxmpp.const import LOCATION_DATA
from nbxmpp.const import AdHocStatus
StanzaHandler = namedtuple('StanzaHandler',
'name callback typ ns xmlns priority')
StanzaHandler.__new__.__defaults__ = ('', '', None, 50)
CommonResult = namedtuple('CommonResult', 'jid')
CommonResult.__new__.__defaults__ = (None,)
InviteData = namedtuple('InviteData',
'muc from_ reason password type continued thread')
DeclineData = namedtuple('DeclineData', 'muc from_ reason')
CaptchaData = namedtuple('CaptchaData', 'form bob_data')
BobData = namedtuple('BobData', 'algo hash_ max_age data cid type')
VoiceRequest = namedtuple('VoiceRequest', 'form jid nick')
MucUserData = namedtuple('MucUserData', 'affiliation jid nick role actor reason')
MucUserData.__new__.__defaults__ = (None, None, None, None, None)
MucDestroyed = namedtuple('MucDestroyed', 'alternate reason password')
MucDestroyed.__new__.__defaults__ = (None, None, None)
MucConfigResult = namedtuple('MucConfigResult', 'jid form')
MucConfigResult.__new__.__defaults__ = (None,)
AffiliationResult = namedtuple('AffiliationResult', 'jid users')
EntityCapsData = namedtuple('EntityCapsData', 'hash node ver')
EntityCapsData.__new__.__defaults__ = (None, None, None)
HTTPAuthData = namedtuple('HTTPAuthData', 'id method url body')
HTTPAuthData.__new__.__defaults__ = (None, None, None, None)
StanzaIDData = namedtuple('StanzaIDData', 'id by')
StanzaIDData.__new__.__defaults__ = (None, None)
PubSubEventData = namedtuple('PubSubEventData', 'node id item data deleted retracted purged')
PubSubEventData.__new__.__defaults__ = (None, None, None, False, False, False)
PubSubConfigResult = namedtuple('PubSubConfigResult', 'jid node form')
PubSubPublishResult = namedtuple('PubSubPublishResult', 'jid node id')
MoodData = namedtuple('MoodData', 'mood text')
ActivityData = namedtuple('ActivityData', 'activity subactivity text')
from nbxmpp.const import AdHocNoteType
from nbxmpp.const import InviteType
from nbxmpp.const import Role
from nbxmpp.const import Affiliation
from nbxmpp.const import AnonymityMode
# TODO: Import it from nbxmpp.modules.dataforms maybe.
Form = Any
@dataclass(frozen=True)
class StanzaHandler:
name: str
callback: Callable[[Any, Any, Any], None]
typ: str = ''
ns: str = ''
xmlns: Optional[str] = None
priority: int = 50
@dataclass(frozen=True)
class CommonResult:
jid: Optional[JID] = None
@dataclass(frozen=True)
class InviteData:
muc: JID
from_: JID
reason: str
password: str
type: InviteType
continued: bool
thread: str
@dataclass(frozen=True)
class DeclineData:
muc: JID
from_: JID
reason: str
@dataclass(frozen=True)
class BobData:
algo: str
hash_: str
max_age: int
data: bytes
cid: str
type: str
@dataclass(frozen=True)
class CaptchaData:
form: Form
bob_data: BobData
@dataclass(frozen=True)
class VoiceRequest:
form: Form
jid: JID
nick: str
@dataclass(frozen=True)
class MucUserData:
affiliation: Affiliation
jid: JID
role: Role
nick: Optional[str] = None
actor: Optional[str] = None
reason: Optional[str] = None
@dataclass(frozen=True)
class MucDestroyed:
alternate: Optional[JID] = None
reason: Optional[str] = None
password: Optional[str] = None
@dataclass(frozen=True)
class MucConfigResult:
jid: JID
form: Optional[Form] = None
@dataclass(frozen=True)
class AffiliationResult:
jid: JID
users: Dict[JID, Dict[str, Optional[str]]]
@dataclass(frozen=True)
class EntityCapsData:
hash: Optional[str] = None
node: Optional[str] = None
ver: Optional[str] = None
@dataclass(frozen=True)
class HTTPAuthData:
id: Optional[str] = None
method: Optional[str] = None
url: Optional[str] = None
body: Optional[str] = None
@dataclass(frozen=True)
class StanzaIDData:
id: str
by: str
@dataclass(frozen=True)
class PubSubEventData:
node: str
id: Optional[str] = None
item: Optional[str] = None
data: Optional[Any] = None
deleted: bool = False
retracted: bool = False
purged: bool = False
@dataclass(frozen=True)
class PubSubConfigResult:
jid: JID
node: str
form: Form
@dataclass(frozen=True)
class PubSubPublishResult:
jid: JID
node: Optional[str] = None
id: Optional[str] = None
@dataclass(frozen=True)
class MoodData:
mood: str
text: Optional[str]
@dataclass(frozen=True)
class ActivityData:
activity: str
subactivity: Optional[str]
text: Optional[str]
LocationData = namedtuple('LocationData', LOCATION_DATA)
LocationData.__new__.__defaults__ = (None,) * len(LocationData._fields)
AvatarMetaData = namedtuple('AvatarMetaData', 'bytes height width id type url')
AvatarMetaData.__new__.__defaults__ = (None,) * len(AvatarMetaData._fields)
AvatarData = namedtuple('AvatarData', 'jid sha data')
AvatarData.__new__.__defaults__ = (None,) * len(AvatarData._fields)
BookmarkData = namedtuple('BookmarkData', 'jid name nick autojoin password')
BookmarkData.__new__.__defaults__ = (None, None, None, None)
BlockingListResult = namedtuple('BlockingListResult', 'blocking_list')
PGPPublicKey = namedtuple('PGPPublicKey', 'jid key date')
PGPKeyMetadata = namedtuple('PGPKeyMetadata', 'jid fingerprint date')
OMEMOMessage = namedtuple('OMEMOMessage', 'sid iv keys payload')
@dataclass(frozen=True)
class AvatarMetaData:
bytes: int
id: str
type: str
height: Optional[int] = None
width: Optional[int] = None
url: Optional[str] = None
@dataclass(frozen=True)
class AvatarData:
jid: JID
sha: Optional[str]
data: bytes
@dataclass(frozen=True)
class BookmarkData:
jid: JID
name: Optional[str] = None
nick: Optional[str] = None
autojoin: bool = False
password: Optional[str] = None
@dataclass(frozen=True)
class BlockingListResult:
blocking_list: List[JID]
@dataclass(frozen=True)
class PGPPublicKey:
jid: JID
key: bytes
date: Optional[datetime] = None
@dataclass(frozen=True)
class PGPKeyMetadata:
jid: JID
fingerprint: str
date: datetime
@dataclass(frozen=True)
class OMEMOMessage:
sid: int
iv: bytes
keys: Dict[int, Tuple[bytes, bool]]
payload: bytes
@dataclass(frozen=True)
class AnnotationNote:
jid: JID
data: Optional[str]
cdate: Optional[datetime] = None
mdate: Optional[datetime] = None
@dataclass(frozen=True)
class EMEData:
name: Optional[str]
namespace: str
@dataclass(frozen=True)
class MuclumbusItem:
jid: JID
name: str
nusers: str
description: str
language: str
is_open: bool
anonymity_mode: AnonymityMode
@dataclass(frozen=True)
class MuclumbusResult:
items: List[MuclumbusItem]
end: bool
first: Optional[str] = None
last: Optional[str] = None
max: Optional[int] = None
@dataclass(frozen=True)
class SoftwareVersionResult:
name: str
version: str
os: Optional[str]
@dataclass(frozen=True)
class AdHocCommandNote:
text: Optional[str]
type: AdHocNoteType = AdHocNoteType.INFO
@dataclass(frozen=True)
class IBBData:
sid: str
block_size: Optional[int] = None
seq: Optional[int] = None
type: Optional[str] = None
data: Optional[bytes] = None
@dataclass(frozen=True)
class DiscoItem:
jid: JID
name: Optional[str] = None
node: Optional[str] = None
@dataclass(frozen=True)
class DiscoItems:
jid: JID
items: List[DiscoItem]
node: Optional[str] = None
@dataclass(frozen=True)
class OOBData:
url: str
desc: Optional[str]
@dataclass(frozen=True)
class CorrectionData:
id: str
@dataclass(frozen=True)
class DisplayMarking:
label: str
fgcolor: Optional[str] = None
bgcolor: Optional[str] = None
@dataclass(frozen=True)
class SecurityLabel:
displaymarking: DisplayMarking
@dataclass(frozen=True)
class RegisterData:
instructions: Optional[str] = None
form: Optional[Form] = None
fields_form: Optional[Form] = None
oob_url: Optional[str] = None
bob_data: Optional[BobData] = None
@dataclass(frozen=True)
class ChangePasswordResult:
successful: bool
form: Optional[Form] = None
@dataclass(frozen=True)
class HTTPUploadData:
put_uri: str
get_uri: str
headers: Dict[str, str]
@dataclass(frozen=True)
class RSMData:
after: Optional[str] = None
before: Optional[str] = None
last: Optional[str] = None
first: Optional[str] = None
first_index: Optional[int] = None
count: Optional[int] = None
max: Optional[int] = None
index: Optional[int] = None
@dataclass(frozen=True)
class MAMQueryData:
jid: JID
rsm: RSMData
complete: bool = False
@dataclass(frozen=True)
class MAMPreferencesData:
default: str
always: List[JID]
never: List[JID]
@dataclass(frozen=True, eq=False)
class DiscoIdentity:
category: str
type: str
name: Optional[str] = None
lang: Optional[str] = None
AnnotationNote = namedtuple('AnnotationNote', 'jid data cdate mdate')
AnnotationNote.__new__.__defaults__ = (None, None)
EMEData = namedtuple('EMEData', 'name namespace')
MuclumbusResult = namedtuple('MuclumbusResult', 'first last max end items')
MuclumbusItem = namedtuple('MuclumbusItem', 'jid name nusers description language is_open anonymity_mode')
SoftwareVersionResult = namedtuple('SoftwareVersionResult', 'name version os')
AdHocCommandNote = namedtuple('AdHocCommandNote', 'text type')
IBBData = namedtuple('IBBData', 'block_size sid seq type data')
IBBData.__new__.__defaults__ = (None, None, None, None, None)
DiscoItems = namedtuple('DiscoItems', 'jid node items')
DiscoItem = namedtuple('DiscoItem', 'jid name node')
DiscoItem.__new__.__defaults__ = (None, None)
OOBData = namedtuple('OOBData', 'url desc')
CorrectionData = namedtuple('CorrectionData', 'id')
DisplayMarking = namedtuple('DisplayMarking', 'label fgcolor bgcolor')
SecurityLabel = namedtuple('SecurityLabel', 'displaymarking')
RegisterData = namedtuple('RegisterData', 'instructions form fields_form oob_url bob_data')
ChangePasswordResult = namedtuple('ChangePasswordResult', 'successful form')
ChangePasswordResult.__new__.__defaults__ = (None,)
HTTPUploadData = namedtuple('HTTPUploadData', 'put_uri get_uri headers')
HTTPUploadData.__new__.__defaults__ = (None,)
def get_node(self):
identity = Node('identity',
attrs={'category': self.category,
'type': self.type})
if self.name is not None:
identity.setAttr('name', self.name)
RSMData = namedtuple('RSMData', 'after before last first first_index count max index')
if self.lang is not None:
identity.setAttr('xml:lang', self.lang)
return identity
MAMQueryData = namedtuple('MAMQueryData', 'jid rsm complete')
def __eq__(self, other):
return str(self) == str(other)
MAMPreferencesData = namedtuple('MAMPreferencesData', 'default always never')
def __ne__(self, other):
return not self.__eq__(other)
def __str__(self):
return '%s/%s/%s/%s' % (self.category,
self.type,
self.lang or '',
self.name or '')
class DiscoInfo(namedtuple('DiscoInfo', 'stanza identities features dataforms timestamp')):
def __hash__(self):
return hash(str(self))
__slots__ = []
def __new__(cls, stanza, identities, features, dataforms, timestamp=None):
return super(DiscoInfo, cls).__new__(cls, stanza, identities,
features, dataforms, timestamp)
@dataclass(frozen=True)
class DiscoInfo:
# TODO: Import it from somewhere maybe.
stanza: Any
identities: List[DiscoIdentity]
features: List[str]
dataforms: List[Form]
timestamp: Optional[datetime] = None
def get_caps_hash(self):
try:
......@@ -387,48 +619,16 @@ class DiscoInfo(namedtuple('DiscoInfo', 'stanza identities features dataforms ti
return None
class DiscoIdentity(namedtuple('DiscoIdentity', 'category type name lang')):
__slots__ = []
def __new__(cls, category, type, name=None, lang=None):
return super(DiscoIdentity, cls).__new__(cls, category, type, name, lang)
def get_node(self):
identity = Node('identity',
attrs={'category': self.category,
'type': self.type})
if self.name is not None:
identity.setAttr('name', self.name)
if self.lang is not None:
identity.setAttr('xml:lang', self.lang)
return identity
def __eq__(self, other):
return str(self) == str(other)
def __ne__(self, other):
return not self.__eq__(other)
def __str__(self):
return '%s/%s/%s/%s' % (self.category,
self.type,
self.lang or '',
self.name or '')
def __hash__(self):
return hash(str(self))
class AdHocCommand(namedtuple('AdHocCommand', 'jid node name sessionid status data actions notes')):
__slots__ = []
def __new__(cls, jid, node, name, sessionid=None, status=None,
data=None, actions=None, notes=None):
return super(AdHocCommand, cls).__new__(cls, jid, node, name, sessionid,
status, data, actions, notes)
@dataclass(frozen=True)
class AdHocCommand:
jid: JID
node: str
name: str
sessionid: Optional[str] = None
status: Optional[AdHocStatus] = None
data: Optional[str] = None
actions: Optional[str] = None
notes: Optional[str] = None
@property
def is_completed(self):
......@@ -439,9 +639,12 @@ class AdHocCommand(namedtuple('AdHocCommand', 'jid node name sessionid status da
return self.status == AdHocStatus.CANCELED
class ProxyData(namedtuple('ProxyData', 'type host username password')):
__slots__ = []
@dataclass(frozen=True)
class ProxyData:
type: str
host: str
username: Optional[str] = None
password: Optional[str] = None
def get_uri(self):
if self.username is not None:
......@@ -456,12 +659,21 @@ class ProxyData(namedtuple('ProxyData', 'type host username password')):
return Gio.SimpleProxyResolver.new(self.get_uri(), None)
class OMEMOBundle(namedtuple('OMEMOBundle', 'spk spk_signature ik otpks')):
@dataclass(frozen=True)
class OMEMOBundle:
spk: str
spk_signature: str
ik: str
otpks: str
def pick_prekey(self):
return random.SystemRandom().choice(self.otpks)
class ChatMarker(namedtuple('ChatMarker', 'type id')):
@dataclass(frozen=True)
class ChatMarker:
type: str
id: str
@property
def is_received(self):
......@@ -627,14 +839,15 @@ class StreamError(CommonError):
raise NotImplementedError
class TuneData(namedtuple('TuneData', 'artist length rating source title track uri')):
__slots__ = []
def __new__(cls, artist=None, length=None, rating=None, source=None,
title=None, track=None, uri=None):
return super(TuneData, cls).__new__(cls, artist, length, rating,
source, title, track, uri)
@dataclass(frozen=True)
class TuneData:
artist: Optional[str] = None
length: Optional[str] = None
rating: Optional[int] = None
source: Optional[str] = None
title: Optional[str] = None
track: Optional[str] = None
uri: Optional[str] = None
@property
def was_removed(self):
......@@ -643,9 +856,13 @@ class TuneData(namedtuple('TuneData', 'artist length rating source title track u
self.track is None)
class MAMData(namedtuple('MAMData', 'id query_id archive namespace timestamp')):
__slots__ = []
@dataclass(frozen=True)
class MAMData:
id: str
query_id: str
archive: str
namespace: str
timestamp: str
@property
def is_ver_1(self):
......@@ -656,9 +873,9 @@ class MAMData(namedtuple('MAMData', 'id query_id archive namespace timestamp')):
return self.namespace == NS_MAM_2
class CarbonData(namedtuple('MAMData', 'type')):
__slots__ = []
@dataclass(frozen=True)
class CarbonData:
type: str