Refactor/remove python38 support

.github/workflows/python-app.yml

@@ -22,7 +22,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: [ 3.8, 3.9, "3.10", "3.11", "3.12" ]
+        python-version: [ 3.9, "3.10", "3.11", "3.12" ]
         os: [ ubuntu-latest, macos-latest, windows-latest ]
 
     steps:

.pre-commit-config.yaml

@@ -5,7 +5,7 @@ repos:
     hooks:
       - id: isort
         args: [ '-m', 'HANGING_INDENT', '-l', '120','--check-only' ]
-        files: \.py$
+        files: .
 
   - repo: https://github.com/pycqa/flake8
     rev: "7.0.0"

misc/plist_sniffer.py

@@ -1,7 +1,7 @@
 import plistlib
 import pprint
 import xml
-from typing import IO, Mapping, Optional, TextIO
+from typing import IO, Optional, TextIO
 
 import click
 from scapy.packet import Packet, Raw
@@ -31,7 +31,7 @@ def process_packet(self, packet: Packet) -> None:
             except xml.parsers.expat.ExpatError:
                 pass
 
-    def report(self, plist: Mapping) -> None:
+    def report(self, plist: dict) -> None:
         try:
             print(plist)
             if self.file is not None:

misc/remotexpc_sniffer.py

@@ -1,6 +1,6 @@
 import logging
 from pprint import pformat
-from typing import List, MutableMapping, Optional
+from typing import Optional
 
 import click
 import coloredlogs
@@ -71,7 +71,7 @@ def add(self, tcp_pkt: TCP) -> bool:
 
 
 class H2Stream(TCPStream):
-    def pop_frames(self) -> List[Frame]:
+    def pop_frames(self) -> list[Frame]:
         """ Pop all available H2Frames """
 
         # If self.data starts with the http/2 magic bytes, pop them off
@@ -96,8 +96,8 @@ def pop_frames(self) -> List[Frame]:
 
 class RemoteXPCSniffer:
     def __init__(self):
-        self._h2_streams: MutableMapping[str, H2Stream] = {}
-        self._previous_frame_data: MutableMapping[str, bytes] = {}
+        self._h2_streams: dict[str, H2Stream] = {}
+        self._previous_frame_data: dict[str, bytes] = {}
 
     def process_packet(self, packet: Packet) -> None:
         if packet.haslayer(TCP) and packet[TCP].payload:

pymobiledevice3/bonjour.py

@@ -2,7 +2,7 @@
 import dataclasses
 import sys
 from socket import AF_INET, AF_INET6, inet_ntop
-from typing import List, Mapping, Optional
+from typing import Optional
 
 from ifaddr import get_adapters
 from zeroconf import IPVersion, ServiceListener, ServiceStateChange, Zeroconf
@@ -22,19 +22,19 @@
 @dataclasses.dataclass
 class BonjourAnswer:
     name: str
-    properties: Mapping[bytes, bytes]
-    ips: List[str]
+    properties: dict[bytes, bytes]
+    ips: list[str]
     port: int
 
 
 class BonjourListener(ServiceListener):
     def __init__(self, ip: str):
         super().__init__()
         self.name: Optional[str] = None
-        self.properties: Mapping[bytes, bytes] = {}
+        self.properties: dict[bytes, bytes] = {}
         self.ip = ip
         self.port: Optional[int] = None
-        self.addresses: List[str] = []
+        self.addresses: list[str] = []
         self.queue: asyncio.Queue = asyncio.Queue()
         self.querying_task: Optional[asyncio.Task] = asyncio.create_task(self.query_addresses())
 
@@ -72,16 +72,16 @@ class BonjourQuery:
     listener: BonjourListener
 
 
-def query_bonjour(service_names: List[str], ip: str) -> BonjourQuery:
+def query_bonjour(service_names: list[str], ip: str) -> BonjourQuery:
     aiozc = AsyncZeroconf(interfaces=[ip])
     listener = BonjourListener(ip)
     service_browser = AsyncServiceBrowser(aiozc.zeroconf, service_names,
                                           handlers=[listener.async_on_service_state_change])
     return BonjourQuery(aiozc, service_browser, listener)
 
 
-async def browse(service_names: List[str], ips: List[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) \
-        -> List[BonjourAnswer]:
+async def browse(service_names: list[str], ips: list[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) \
+        -> list[BonjourAnswer]:
     try:
         bonjour_queries = [query_bonjour(service_names, adapter) for adapter in ips]
     except ValueError as e:
@@ -104,11 +104,11 @@ async def browse(service_names: List[str], ips: List[str], timeout: float = DEFA
     return answers
 
 
-async def browse_ipv6(service_names: List[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_ipv6(service_names: list[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse(service_names, OSUTILS.get_ipv6_ips(), timeout=timeout)
 
 
-def get_ipv4_addresses() -> List[str]:
+def get_ipv4_addresses() -> list[str]:
     ips = []
     for adapter in get_adapters():
         if adapter.nice_name.startswith('tun'):
@@ -123,21 +123,21 @@ def get_ipv4_addresses() -> List[str]:
     return ips
 
 
-async def browse_ipv4(service_names: List[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_ipv4(service_names: list[str], timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse(service_names, get_ipv4_addresses(), timeout=timeout)
 
 
-async def browse_remoted(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_remoted(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse_ipv6(REMOTED_SERVICE_NAMES, timeout=timeout)
 
 
-async def browse_mobdev2(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_mobdev2(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse(MOBDEV2_SERVICE_NAMES, get_ipv4_addresses() + OSUTILS.get_ipv6_ips(), timeout=timeout)
 
 
-async def browse_remotepairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_remotepairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse_ipv4(REMOTEPAIRING_SERVICE_NAMES, timeout=timeout)
 
 
-async def browse_remotepairing_manual_pairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[BonjourAnswer]:
+async def browse_remotepairing_manual_pairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[BonjourAnswer]:
     return await browse_ipv4(REMOTEPAIRING_MANUAL_PAIRING_SERVICE_NAMES, timeout=timeout)

pymobiledevice3/cli/apps.py

@@ -1,5 +1,3 @@
-from typing import List
-
 import click
 
 from pymobiledevice3.cli.cli_common import Command, print_json
@@ -33,7 +31,7 @@ def apps_list(service_provider: LockdownServiceProvider, app_type: str, calculat
 @apps.command('query', cls=Command)
 @click.argument('bundle_identifiers', nargs=-1)
 @click.option('--calculate-sizes/--no-calculate-size', default=False)
-def apps_query(service_provider: LockdownServiceProvider, bundle_identifiers: List[str], calculate_sizes: bool) -> None:
+def apps_query(service_provider: LockdownServiceProvider, bundle_identifiers: list[str], calculate_sizes: bool) -> None:
     """ query installed apps """
     print_json(InstallationProxyService(lockdown=service_provider)
                .get_apps(calculate_sizes=calculate_sizes, bundle_identifiers=bundle_identifiers))

pymobiledevice3/cli/cli_common.py

@@ -6,7 +6,7 @@
 import sys
 import uuid
 from functools import wraps
-from typing import Any, Callable, List, Mapping, Optional, Tuple
+from typing import Any, Callable, Optional
 
 import click
 import coloredlogs
@@ -126,7 +126,7 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def prompt_selection(choices: List[Any], message: str, idx: bool = False) -> Any:
+def prompt_selection(choices: list[Any], message: str, idx: bool = False) -> Any:
     question = [inquirer3.List('selection', message=message, choices=choices, carousel=True)]
     try:
         result = inquirer3.prompt(question, theme=GreenPassion(), raise_keyboard_interrupt=True)
@@ -135,12 +135,12 @@ def prompt_selection(choices: List[Any], message: str, idx: bool = False) -> Any
     return result['selection'] if not idx else choices.index(result['selection'])
 
 
-def prompt_device_list(device_list: List):
+def prompt_device_list(device_list: list):
     return prompt_selection(device_list, 'Choose device')
 
 
 def choose_service_provider(callback: Callable):
-    def wrap_callback_calling(**kwargs: Mapping):
+    def wrap_callback_calling(**kwargs: dict) -> None:
         service_provider = None
         lockdown_service_provider = kwargs.pop('lockdown_service_provider', None)
         rsd_service_provider_manually = kwargs.pop('rsd_service_provider_manually', None)
@@ -229,7 +229,7 @@ def __init__(self, *args, **kwargs):
                            f'This option may also be transferred as an environment variable: {TUNNEL_ENV_VAR}')
         ]
 
-    def rsd(self, ctx, param: str, value: Optional[Tuple[str, int]]) -> Optional[RemoteServiceDiscoveryService]:
+    def rsd(self, ctx, param: str, value: Optional[tuple[str, int]]) -> Optional[RemoteServiceDiscoveryService]:
         if value is not None:
             rsd = RemoteServiceDiscoveryService(value)
             asyncio.run(rsd.connect(), debug=True)

pymobiledevice3/cli/developer.py

@@ -11,7 +11,7 @@
 from dataclasses import asdict
 from datetime import datetime
 from pathlib import Path
-from typing import IO, List, Optional, Tuple
+from typing import IO, Optional
 
 import click
 from click.exceptions import MissingParameter, UsageError
@@ -165,7 +165,7 @@ def process_id_for_bundle_id(service_provider: LockdownServiceProvider, app_bund
 
 def get_matching_processes(service_provider: LockdownServiceProvider, name: Optional[str] = None,
                            bundle_identifier: Optional[str] = None) \
-        -> List[MatchedProcessByPid]:
+        -> list[MatchedProcessByPid]:
     result = []
     with DvtSecureSocketProxyService(lockdown=service_provider) as dvt:
         device_info = DeviceInfo(dvt)
@@ -331,7 +331,7 @@ def sysmon_process_monitor(service_provider: LockdownClient, threshold):
 @sysmon_process.command('single', cls=Command)
 @click.option('-a', '--attributes', multiple=True,
               help='filter processes by given attribute value given as key=value')
-def sysmon_process_single(service_provider: LockdownClient, attributes: List[str]):
+def sysmon_process_single(service_provider: LockdownClient, attributes: list[str]):
     """ show a single snapshot of currently running processes. """
 
     count = 0
@@ -403,7 +403,7 @@ def core_profile_session():
                                help='Events subclass filter. Omit for all.')
 
 
-def parse_filters(subclasses: List[int], classes: List[int]):
+def parse_filters(subclasses: list[int], classes: list[int]):
     if not subclasses and not classes:
         return None
     parsed = set()
@@ -1096,8 +1096,8 @@ def core_device_read_file(
 
 
 async def core_device_list_launch_application_task(
-        service_provider: RemoteServiceDiscoveryService, bundle_identifier: str, argument: List[str],
-        kill_existing: bool, suspended: bool, env: List[Tuple[str, str]]) -> None:
+        service_provider: RemoteServiceDiscoveryService, bundle_identifier: str, argument: list[str],
+        kill_existing: bool, suspended: bool, env: list[tuple[str, str]]) -> None:
     async with AppServiceService(service_provider) as app_service:
         print_json(await app_service.launch_application(bundle_identifier, argument, kill_existing,
                                                         suspended, dict(env)))
@@ -1112,8 +1112,8 @@ async def core_device_list_launch_application_task(
 @click.option('--env', multiple=True, type=click.Tuple((str, str)),
               help='Environment variables to pass to process given as a list of key value')
 def core_device_launch_application(
-        service_provider: RemoteServiceDiscoveryService, bundle_identifier: str, argument: Tuple[str],
-        kill_existing: bool, suspended: bool, env: List[Tuple[str, str]]) -> None:
+        service_provider: RemoteServiceDiscoveryService, bundle_identifier: str, argument: tuple[str],
+        kill_existing: bool, suspended: bool, env: list[tuple[str, str]]) -> None:
     """ Launch application """
     asyncio.run(
         core_device_list_launch_application_task(
@@ -1180,15 +1180,15 @@ def core_device_get_display_info(service_provider: RemoteServiceDiscoveryService
     asyncio.run(core_device_get_display_info_task(service_provider))
 
 
-async def core_device_query_mobilegestalt_task(service_provider: RemoteServiceDiscoveryService, key: List[str]) -> None:
+async def core_device_query_mobilegestalt_task(service_provider: RemoteServiceDiscoveryService, key: list[str]) -> None:
     """ Query MobileGestalt """
     async with DeviceInfoService(service_provider) as app_service:
         print_json(await app_service.query_mobilegestalt(key))
 
 
 @core_device.command('query-mobilegestalt', cls=RSDCommand)
 @click.argument('key', nargs=-1, type=click.STRING)
-def core_device_query_mobilegestalt(service_provider: RemoteServiceDiscoveryService, key: Tuple[str]) -> None:
+def core_device_query_mobilegestalt(service_provider: RemoteServiceDiscoveryService, key: tuple[str]) -> None:
     """ Query MobileGestalt """
     asyncio.run(core_device_query_mobilegestalt_task(service_provider, list(key)))
 

pymobiledevice3/cli/profile.py

@@ -2,7 +2,7 @@
 import plistlib
 import tempfile
 from pathlib import Path
-from typing import IO, List, Optional
+from typing import IO, Optional
 
 import click
 
@@ -36,7 +36,7 @@ def profile_list(service_provider: LockdownClient):
 @profile_group.command('install', cls=Command)
 @click.option('--keybag', type=click.Path(file_okay=True, dir_okay=False, exists=True))
 @click.argument('profiles', nargs=-1, type=click.File('rb'))
-def profile_install(service_provider: LockdownServiceProvider, keybag: Optional[str], profiles: List[IO]) -> None:
+def profile_install(service_provider: LockdownServiceProvider, keybag: Optional[str], profiles: list[IO]) -> None:
     """
     Install given profiles
 
@@ -66,7 +66,7 @@ def profile_cloud_configuration(service_provider: LockdownServiceProvider, confi
 
 @profile_group.command('store', cls=Command)
 @click.argument('profiles', nargs=-1, type=click.File('rb'))
-def profile_store(service_provider: LockdownServiceProvider, profiles: List[IO]) -> None:
+def profile_store(service_provider: LockdownServiceProvider, profiles: list[IO]) -> None:
     """ Store a profile """
     service = MobileConfigService(lockdown=service_provider)
     for profile in profiles:

pymobiledevice3/cli/remote.py

@@ -4,7 +4,7 @@
 import sys
 import tempfile
 from functools import partial
-from typing import List, Mapping, Optional, TextIO
+from typing import Optional, TextIO
 
 import click
 
@@ -25,7 +25,7 @@
 logger = logging.getLogger(__name__)
 
 
-async def browse_rsd(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[Mapping]:
+async def browse_rsd(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[dict]:
     devices = []
     for rsd in await get_rsds(timeout):
         devices.append({'address': rsd.service.address[0],
@@ -36,7 +36,7 @@ async def browse_rsd(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[Mapping]:
     return devices
 
 
-async def browse_remotepairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> List[Mapping]:
+async def browse_remotepairing(timeout: float = DEFAULT_BONJOUR_TIMEOUT) -> list[dict]:
     devices = []
     for remotepairing in await get_remote_pairing_tunnel_services(timeout):
         devices.append({'address': remotepairing.hostname,
@@ -211,7 +211,7 @@ async def start_remote_pair_task(device_name: str) -> None:
     if start_tunnel is None:
         raise NotImplementedError('failed to start the tunnel on your platform')
 
-    devices: List[RemotePairingManualPairingDevice] = []
+    devices: list[RemotePairingManualPairingDevice] = []
     for answer in await browse_remotepairing_manual_pairing():
         current_device_name = answer.properties[b'name'].decode()
 

pymobiledevice3/cli/restore.py

@@ -5,8 +5,9 @@
 import plistlib
 import tempfile
 import traceback
+from collections.abc import Generator
 from pathlib import Path
-from typing import IO, Generator, Optional, Union
+from typing import IO, Optional, Union
 from zipfile import ZipFile
 
 import click

pymobiledevice3/cli/syslog.py

@@ -2,7 +2,7 @@
 import os
 import posixpath
 import re
-from typing import List, Optional, TextIO
+from typing import Optional, TextIO
 
 import click
 
@@ -82,8 +82,8 @@ def format_line(color, pid, syslog_entry, include_label):
 
 def syslog_live(
         service_provider: LockdownServiceProvider, out: Optional[TextIO], pid: Optional[int],
-        process_name: Optional[str], match: List[str], match_insensitive: List[str], include_label: bool,
-        regex: List[str], insensitive_regex: List[str]) -> None:
+        process_name: Optional[str], match: list[str], match_insensitive: list[str], include_label: bool,
+        regex: list[str], insensitive_regex: list[str]) -> None:
     match_regex = [re.compile(f'.*({r}).*', re.DOTALL) for r in regex]
     match_regex += [re.compile(f'.*({r}).*', re.IGNORECASE | re.DOTALL) for r in insensitive_regex]
 
@@ -158,8 +158,8 @@ def replace(m):
 @click.option('-ei', '--insensitive-regex', multiple=True, help='filter only lines matching given regex (insensitive)')
 def cli_syslog_live(
         service_provider: LockdownServiceProvider, out: Optional[TextIO], pid: Optional[int],
-        process_name: Optional[str], match: List[str], match_insensitive: List[str], include_label: bool,
-        regex: List[str], insensitive_regex: List[str]) -> None:
+        process_name: Optional[str], match: list[str], match_insensitive: list[str], include_label: bool,
+        regex: list[str], insensitive_regex: list[str]) -> None:
     """ view live syslog lines """
 
     syslog_live(service_provider, out, pid, process_name, match, match_insensitive, include_label, regex,