refactor(deltachat-rpc-client): use list, set and tuple instead of typing

`typing.List` is deprecated according to https://docs.python.org/3/library/typing.html#typing.List
Similar for `Set` and `Dict`.

`from __future__ import annotations` is for compatibility with Python 3.7.
This commit is contained in:
link2xt
2024-04-08 21:57:37 +00:00
parent 27d2b12e8d
commit 9aa4c0e56b
8 changed files with 54 additions and 42 deletions

View File

@@ -1,5 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import TYPE_CHECKING, List, Optional, Tuple, Union from typing import TYPE_CHECKING, Optional, Union
from warnings import warn from warnings import warn
from ._utils import AttrDict, futuremethod from ._utils import AttrDict, futuremethod
@@ -126,7 +128,7 @@ class Account:
contact_id = self._rpc.lookup_contact_id_by_addr(self.id, address) contact_id = self._rpc.lookup_contact_id_by_addr(self.id, address)
return contact_id and Contact(self, contact_id) return contact_id and Contact(self, contact_id)
def get_blocked_contacts(self) -> List[AttrDict]: def get_blocked_contacts(self) -> list[AttrDict]:
"""Return a list with snapshots of all blocked contacts.""" """Return a list with snapshots of all blocked contacts."""
contacts = self._rpc.get_blocked_contacts(self.id) contacts = self._rpc.get_blocked_contacts(self.id)
return [AttrDict(contact=Contact(self, contact["id"]), **contact) for contact in contacts] return [AttrDict(contact=Contact(self, contact["id"]), **contact) for contact in contacts]
@@ -151,7 +153,7 @@ class Account:
with_self: bool = False, with_self: bool = False,
verified_only: bool = False, verified_only: bool = False,
snapshot: bool = False, snapshot: bool = False,
) -> Union[List[Contact], List[AttrDict]]: ) -> Union[list[Contact], list[AttrDict]]:
"""Get a filtered list of contacts. """Get a filtered list of contacts.
:param query: if a string is specified, only return contacts :param query: if a string is specified, only return contacts
@@ -186,7 +188,7 @@ class Account:
no_specials: bool = False, no_specials: bool = False,
alldone_hint: bool = False, alldone_hint: bool = False,
snapshot: bool = False, snapshot: bool = False,
) -> Union[List[Chat], List[AttrDict]]: ) -> Union[list[Chat], list[AttrDict]]:
"""Return list of chats. """Return list of chats.
:param query: if a string is specified only chats matching this query are returned. :param query: if a string is specified only chats matching this query are returned.
@@ -244,7 +246,7 @@ class Account:
""" """
return Chat(self, self._rpc.secure_join(self.id, qrdata)) return Chat(self, self._rpc.secure_join(self.id, qrdata))
def get_qr_code(self) -> Tuple[str, str]: def get_qr_code(self) -> tuple[str, str]:
"""Get Setup-Contact QR Code text and SVG data. """Get Setup-Contact QR Code text and SVG data.
this data needs to be transferred to another Delta Chat account this data needs to be transferred to another Delta Chat account
@@ -256,15 +258,15 @@ class Account:
"""Return the Message instance with the given ID.""" """Return the Message instance with the given ID."""
return Message(self, msg_id) return Message(self, msg_id)
def mark_seen_messages(self, messages: List[Message]) -> None: def mark_seen_messages(self, messages: list[Message]) -> None:
"""Mark the given set of messages as seen.""" """Mark the given set of messages as seen."""
self._rpc.markseen_msgs(self.id, [msg.id for msg in messages]) self._rpc.markseen_msgs(self.id, [msg.id for msg in messages])
def delete_messages(self, messages: List[Message]) -> None: def delete_messages(self, messages: list[Message]) -> None:
"""Delete messages (local and remote).""" """Delete messages (local and remote)."""
self._rpc.delete_messages(self.id, [msg.id for msg in messages]) self._rpc.delete_messages(self.id, [msg.id for msg in messages])
def get_fresh_messages(self) -> List[Message]: def get_fresh_messages(self) -> list[Message]:
"""Return the list of fresh messages, newest messages first. """Return the list of fresh messages, newest messages first.
This call is intended for displaying notifications. This call is intended for displaying notifications.
@@ -274,12 +276,12 @@ class Account:
fresh_msg_ids = self._rpc.get_fresh_msgs(self.id) fresh_msg_ids = self._rpc.get_fresh_msgs(self.id)
return [Message(self, msg_id) for msg_id in fresh_msg_ids] return [Message(self, msg_id) for msg_id in fresh_msg_ids]
def get_next_messages(self) -> List[Message]: def get_next_messages(self) -> list[Message]:
"""Return a list of next messages.""" """Return a list of next messages."""
next_msg_ids = self._rpc.get_next_msgs(self.id) next_msg_ids = self._rpc.get_next_msgs(self.id)
return [Message(self, msg_id) for msg_id in next_msg_ids] return [Message(self, msg_id) for msg_id in next_msg_ids]
def wait_next_messages(self) -> List[Message]: def wait_next_messages(self) -> list[Message]:
"""Wait for new messages and return a list of them.""" """Wait for new messages and return a list of them."""
next_msg_ids = self._rpc.wait_next_msgs(self.id) next_msg_ids = self._rpc.wait_next_msgs(self.id)
return [Message(self, msg_id) for msg_id in next_msg_ids] return [Message(self, msg_id) for msg_id in next_msg_ids]
@@ -309,7 +311,7 @@ class Account:
if event.kind == EventType.REACTIONS_CHANGED: if event.kind == EventType.REACTIONS_CHANGED:
return event return event
def get_fresh_messages_in_arrival_order(self) -> List[Message]: def get_fresh_messages_in_arrival_order(self) -> list[Message]:
"""Return fresh messages list sorted in the order of their arrival, with ascending IDs.""" """Return fresh messages list sorted in the order of their arrival, with ascending IDs."""
warn( warn(
"get_fresh_messages_in_arrival_order is deprecated, use get_next_messages instead.", "get_fresh_messages_in_arrival_order is deprecated, use get_next_messages instead.",

View File

@@ -1,6 +1,8 @@
from __future__ import annotations
import calendar import calendar
from dataclasses import dataclass from dataclasses import dataclass
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union from typing import TYPE_CHECKING, Optional, Union
from ._utils import AttrDict from ._utils import AttrDict
from .const import ChatVisibility, ViewType from .const import ChatVisibility, ViewType
@@ -93,7 +95,7 @@ class Chat:
"""Return encryption info for this chat.""" """Return encryption info for this chat."""
return self._rpc.get_chat_encryption_info(self.account.id, self.id) return self._rpc.get_chat_encryption_info(self.account.id, self.id)
def get_qr_code(self) -> Tuple[str, str]: def get_qr_code(self) -> tuple[str, str]:
"""Get Join-Group QR code text and SVG data.""" """Get Join-Group QR code text and SVG data."""
return self._rpc.get_chat_securejoin_qr_code_svg(self.account.id, self.id) return self._rpc.get_chat_securejoin_qr_code_svg(self.account.id, self.id)
@@ -117,7 +119,7 @@ class Chat:
html: Optional[str] = None, html: Optional[str] = None,
viewtype: Optional[ViewType] = None, viewtype: Optional[ViewType] = None,
file: Optional[str] = None, file: Optional[str] = None,
location: Optional[Tuple[float, float]] = None, location: Optional[tuple[float, float]] = None,
override_sender_name: Optional[str] = None, override_sender_name: Optional[str] = None,
quoted_msg: Optional[Union[int, Message]] = None, quoted_msg: Optional[Union[int, Message]] = None,
) -> Message: ) -> Message:
@@ -156,7 +158,7 @@ class Chat:
msg_id = self._rpc.send_sticker(self.account.id, self.id, path) msg_id = self._rpc.send_sticker(self.account.id, self.id, path)
return Message(self.account, msg_id) return Message(self.account, msg_id)
def forward_messages(self, messages: List[Message]) -> None: def forward_messages(self, messages: list[Message]) -> None:
"""Forward a list of messages to this chat.""" """Forward a list of messages to this chat."""
msg_ids = [msg.id for msg in messages] msg_ids = [msg.id for msg in messages]
self._rpc.forward_messages(self.account.id, msg_ids, self.id) self._rpc.forward_messages(self.account.id, msg_ids, self.id)
@@ -188,7 +190,7 @@ class Chat:
snapshot["message"] = Message(self.account, snapshot.id) snapshot["message"] = Message(self.account, snapshot.id)
return snapshot return snapshot
def get_messages(self, info_only: bool = False, add_daymarker: bool = False) -> List[Message]: def get_messages(self, info_only: bool = False, add_daymarker: bool = False) -> list[Message]:
"""get the list of messages in this chat.""" """get the list of messages in this chat."""
msgs = self._rpc.get_message_ids(self.account.id, self.id, info_only, add_daymarker) msgs = self._rpc.get_message_ids(self.account.id, self.id, info_only, add_daymarker)
return [Message(self.account, msg_id) for msg_id in msgs] return [Message(self.account, msg_id) for msg_id in msgs]
@@ -223,7 +225,7 @@ class Chat:
contact_id = cnt contact_id = cnt
self._rpc.remove_contact_from_chat(self.account.id, self.id, contact_id) self._rpc.remove_contact_from_chat(self.account.id, self.id, contact_id)
def get_contacts(self) -> List[Contact]: def get_contacts(self) -> list[Contact]:
"""Get the contacts belonging to this chat. """Get the contacts belonging to this chat.
For single/direct chats self-address is not included. For single/direct chats self-address is not included.
@@ -247,7 +249,7 @@ class Chat:
contact: Optional[Contact] = None, contact: Optional[Contact] = None,
timestamp_from: Optional["datetime"] = None, timestamp_from: Optional["datetime"] = None,
timestamp_to: Optional["datetime"] = None, timestamp_to: Optional["datetime"] = None,
) -> List[AttrDict]: ) -> list[AttrDict]:
"""Get list of location snapshots for the given contact in the given timespan.""" """Get list of location snapshots for the given contact in the given timespan."""
time_from = calendar.timegm(timestamp_from.utctimetuple()) if timestamp_from else 0 time_from = calendar.timegm(timestamp_from.utctimetuple()) if timestamp_from else 0
time_to = calendar.timegm(timestamp_to.utctimetuple()) if timestamp_to else 0 time_to = calendar.timegm(timestamp_to.utctimetuple()) if timestamp_to else 0
@@ -255,7 +257,7 @@ class Chat:
result = self._rpc.get_locations(self.account.id, self.id, contact_id, time_from, time_to) result = self._rpc.get_locations(self.account.id, self.id, contact_id, time_from, time_to)
locations = [] locations = []
contacts: Dict[int, Contact] = {} contacts: dict[int, Contact] = {}
for loc in result: for loc in result:
location = AttrDict(loc) location = AttrDict(loc)
location["chat"] = self location["chat"] = self

View File

@@ -1,14 +1,13 @@
"""Event loop implementations offering high level event handling/hooking.""" """Event loop implementations offering high level event handling/hooking."""
from __future__ import annotations
import logging import logging
from typing import ( from typing import (
TYPE_CHECKING, TYPE_CHECKING,
Callable, Callable,
Dict,
Iterable, Iterable,
Optional, Optional,
Set,
Tuple,
Type, Type,
Union, Union,
) )
@@ -39,16 +38,16 @@ class Client:
def __init__( def __init__(
self, self,
account: "Account", account: "Account",
hooks: Optional[Iterable[Tuple[Callable, Union[type, EventFilter]]]] = None, hooks: Optional[Iterable[tuple[Callable, Union[type, EventFilter]]]] = None,
logger: Optional[logging.Logger] = None, logger: Optional[logging.Logger] = None,
) -> None: ) -> None:
self.account = account self.account = account
self.logger = logger or logging self.logger = logger or logging
self._hooks: Dict[type, Set[tuple]] = {} self._hooks: dict[type, set[tuple]] = {}
self._should_process_messages = 0 self._should_process_messages = 0
self.add_hooks(hooks or []) self.add_hooks(hooks or [])
def add_hooks(self, hooks: Iterable[Tuple[Callable, Union[type, EventFilter]]]) -> None: def add_hooks(self, hooks: Iterable[tuple[Callable, Union[type, EventFilter]]]) -> None:
for hook, event in hooks: for hook, event in hooks:
self.add_hook(hook, event) self.add_hook(hook, event)

View File

@@ -1,4 +1,6 @@
from typing import TYPE_CHECKING, Dict, List from __future__ import annotations
from typing import TYPE_CHECKING
from ._utils import AttrDict from ._utils import AttrDict
from .account import Account from .account import Account
@@ -21,7 +23,7 @@ class DeltaChat:
account_id = self.rpc.add_account() account_id = self.rpc.add_account()
return Account(self, account_id) return Account(self, account_id)
def get_all_accounts(self) -> List[Account]: def get_all_accounts(self) -> list[Account]:
"""Return a list of all available accounts.""" """Return a list of all available accounts."""
account_ids = self.rpc.get_all_account_ids() account_ids = self.rpc.get_all_account_ids()
return [Account(self, account_id) for account_id in account_ids] return [Account(self, account_id) for account_id in account_ids]
@@ -44,6 +46,6 @@ class DeltaChat:
"""Get information about the Delta Chat core in this system.""" """Get information about the Delta Chat core in this system."""
return AttrDict(self.rpc.get_system_info()) return AttrDict(self.rpc.get_system_info())
def set_translations(self, translations: Dict[str, str]) -> None: def set_translations(self, translations: dict[str, str]) -> None:
"""Set stock translation strings.""" """Set stock translation strings."""
self.rpc.set_stock_strings(translations) self.rpc.set_stock_strings(translations)

View File

@@ -2,12 +2,13 @@
Internal Python-level IMAP handling used by the tests. Internal Python-level IMAP handling used by the tests.
""" """
from __future__ import annotations
import imaplib import imaplib
import io import io
import pathlib import pathlib
import ssl import ssl
from contextlib import contextmanager from contextlib import contextmanager
from typing import List
from imap_tools import ( from imap_tools import (
AND, AND,
@@ -87,7 +88,7 @@ class DirectImap:
return self.select_folder(foldername) return self.select_folder(foldername)
return None return None
def list_folders(self) -> List[str]: def list_folders(self) -> list[str]:
"""return list of all existing folder names.""" """return list of all existing folder names."""
assert not self._idling assert not self._idling
return [folder.name for folder in self.conn.folder.list()] return [folder.name for folder in self.conn.folder.list()]
@@ -102,11 +103,11 @@ class DirectImap:
if expunge: if expunge:
self.conn.expunge() self.conn.expunge()
def get_all_messages(self) -> List[MailMessage]: def get_all_messages(self) -> list[MailMessage]:
assert not self._idling assert not self._idling
return list(self.conn.fetch()) return list(self.conn.fetch())
def get_unread_messages(self) -> List[str]: def get_unread_messages(self) -> list[str]:
assert not self._idling assert not self._idling
return [msg.uid for msg in self.conn.fetch(AND(seen=False))] return [msg.uid for msg in self.conn.fetch(AND(seen=False))]
@@ -198,7 +199,7 @@ class IdleManager:
self.direct_imap.conn.fetch("1:*") self.direct_imap.conn.fetch("1:*")
self.direct_imap.conn.idle.start() self.direct_imap.conn.idle.start()
def check(self, timeout=None) -> List[bytes]: def check(self, timeout=None) -> list[bytes]:
"""(blocking) wait for next idle message from server.""" """(blocking) wait for next idle message from server."""
self.log("imap-direct: calling idle_check") self.log("imap-direct: calling idle_check")
res = self.direct_imap.conn.idle.poll(timeout=timeout) res = self.direct_imap.conn.idle.poll(timeout=timeout)

View File

@@ -1,8 +1,10 @@
"""High-level classes for event processing and filtering.""" """High-level classes for event processing and filtering."""
from __future__ import annotations
import re import re
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Callable, Iterable, Iterator, Optional, Set, Tuple, Union from typing import TYPE_CHECKING, Callable, Iterable, Iterator, Optional, Union
from .const import EventType from .const import EventType
@@ -263,9 +265,9 @@ class HookCollection:
""" """
def __init__(self) -> None: def __init__(self) -> None:
self._hooks: Set[Tuple[Callable, Union[type, EventFilter]]] = set() self._hooks: set[tuple[Callable, Union[type, EventFilter]]] = set()
def __iter__(self) -> Iterator[Tuple[Callable, Union[type, EventFilter]]]: def __iter__(self) -> Iterator[tuple[Callable, Union[type, EventFilter]]]:
return iter(self._hooks) return iter(self._hooks)
def on(self, event: Union[type, EventFilter]) -> Callable: # noqa def on(self, event: Union[type, EventFilter]) -> Callable: # noqa

View File

@@ -1,6 +1,8 @@
from __future__ import annotations
import os import os
import random import random
from typing import AsyncGenerator, List, Optional from typing import AsyncGenerator, Optional
import pytest import pytest
@@ -57,7 +59,7 @@ class ACFactory:
account.bring_online() account.bring_online()
return account return account
def get_online_accounts(self, num: int) -> List[Account]: def get_online_accounts(self, num: int) -> list[Account]:
futures = [self.get_online_account.future() for _ in range(num)] futures = [self.get_online_account.future() for _ in range(num)]
return [f() for f in futures] return [f() for f in futures]

View File

@@ -1,3 +1,5 @@
from __future__ import annotations
import itertools import itertools
import json import json
import logging import logging
@@ -6,7 +8,7 @@ import subprocess
import sys import sys
from queue import Queue from queue import Queue
from threading import Event, Thread from threading import Event, Thread
from typing import Any, Dict, Iterator, Optional from typing import Any, Iterator, Optional
class JsonRpcError(Exception): class JsonRpcError(Exception):
@@ -67,11 +69,11 @@ class Rpc:
self._kwargs = kwargs self._kwargs = kwargs
self.process: subprocess.Popen self.process: subprocess.Popen
self.id_iterator: Iterator[int] self.id_iterator: Iterator[int]
self.event_queues: Dict[int, Queue] self.event_queues: dict[int, Queue]
# Map from request ID to `threading.Event`. # Map from request ID to `threading.Event`.
self.request_events: Dict[int, Event] self.request_events: dict[int, Event]
# Map from request ID to the result. # Map from request ID to the result.
self.request_results: Dict[int, Any] self.request_results: dict[int, Any]
self.request_queue: Queue[Any] self.request_queue: Queue[Any]
self.closing: bool self.closing: bool
self.reader_thread: Thread self.reader_thread: Thread