pep/slixmpp-omemo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

xep_0384: Use Optional instead of Union[None, T]

Browse source 
Signed-off-by: Maxime “pep” Buquet <pep@bouah.net>
This commit is contained in:
Maxime “pep” Buquet 2018-12-30 23:37:28 +01:00
parent 46a7e414a4
commit 2880aef846
2 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
plugin.py
View file

@ -218,7 +218,7 @@ class XEP_0384(BasePlugin):
iq = self._generate_bundle_iq() iq = self._generate_bundle_iq()
await iq.send() await iq.send()
async def _fetch_bundle(self, jid: str, device_id: int) -> Union[None, ExtendedPublicBundle]: async def _fetch_bundle(self, jid: str, device_id: int) -> Optional[ExtendedPublicBundle]:
node = '%s:%d' % (OMEMO_BUNDLES_NS, device_id) node = '%s:%d' % (OMEMO_BUNDLES_NS, device_id)
try: try:
iq = await self.xmpp['xep_0060'].get_items(jid, node) iq = await self.xmpp['xep_0060'].get_items(jid, node)
@ -357,7 +357,7 @@ class XEP_0384(BasePlugin):
recipients = [jid.bare for jid in recipients] recipients = [jid.bare for jid in recipients]
bundles = {} # type: Dict[str, Dict[int, ExtendedPublicBundle]] bundles = {} # type: Dict[str, Dict[int, ExtendedPublicBundle]]
old_errors = None # type: Union[None, List[Tuple[Exception, Any, Any]]] old_errors = None # type: Optional[List[Tuple[Exception, Any, Any]]]
while True: while True:
# Try to encrypt and resolve errors until there is no error at all # Try to encrypt and resolve errors until there is no error at all
# or if we hit the same set of errors. # or if we hit the same set of errors.

10
storage.py
View file

@ -8,7 +8,7 @@
import os import os
import copy import copy
import json import json
from typing import Any, Dict, List, Set, Union from typing import Any, Dict, List, Optional, Set, Union
import omemo import omemo
@ -17,7 +17,7 @@ class SyncFileStorage(omemo.Storage):
def __init__(self, storage_dir: str) -> None: def __init__(self, storage_dir: str) -> None:
self.storage_dir = storage_dir self.storage_dir = storage_dir
self.__state = None self.__state = None
self.__own_data = None # type: Union[None, Dict[str, Union[str, int]]] self.__own_data = None # type: Optional[Dict[str, Union[str, int]]]
self.__sessions = {} # type: Dict[str, Dict[int, Any]] self.__sessions = {} # type: Dict[str, Dict[int, Any]]
self.__devices = {} # type: Dict[str, Dict[str, Union[List[int], Dict[int, int]]]] self.__devices = {} # type: Dict[str, Dict[str, Union[List[int], Dict[int, int]]]]
self.__trust = {} # type: Dict[str, Dict[int, Dict[str, Any]]] self.__trust = {} # type: Dict[str, Dict[int, Dict[str, Any]]]
@ -92,7 +92,7 @@ class SyncFileStorage(omemo.Storage):
filepath = os.path.join(self.storage_dir, 'sessions.json') filepath = os.path.join(self.storage_dir, 'sessions.json')
os.remove(filepath) os.remove(filepath)
def loadActiveDevices(self, _callback, bare_jid: str) -> Union[None, List[int]]: def loadActiveDevices(self, _callback, bare_jid: str) -> Optional[List[int]]:
if not self.__devices: if not self.__devices:
try: try:
filepath = os.path.join(self.storage_dir, 'devices.json') filepath = os.path.join(self.storage_dir, 'devices.json')
@ -111,7 +111,7 @@ class SyncFileStorage(omemo.Storage):
with open(filepath, 'w') as f: with open(filepath, 'w') as f:
json.dump(self.__devices, f) json.dump(self.__devices, f)
def loadInactiveDevices(self, _callback, bare_jid: str) -> Union[None, Dict[int, int]]: def loadInactiveDevices(self, _callback, bare_jid: str) -> Optional[Dict[int, int]]:
if not self.__devices: if not self.__devices:
try: try:
filepath = os.path.join(self.storage_dir, 'devices.json') filepath = os.path.join(self.storage_dir, 'devices.json')
@ -138,7 +138,7 @@ class SyncFileStorage(omemo.Storage):
with open(filepath, 'w') as f: with open(filepath, 'w') as f:
json.dump(self.__trust, f) json.dump(self.__trust, f)
def loadTrust(self, _callback, bare_jid: str, device_id: int) -> Union[None, Dict[str, Any]]: def loadTrust(self, _callback, bare_jid: str, device_id: int) -> Optional[Dict[str, Any]]:
if not self.__trust: if not self.__trust:
try: try:
filepath = os.path.join(self.storage_dir, 'trust.json') filepath = os.path.join(self.storage_dir, 'trust.json')