diff --git a/actions/actions b/actions/actions index e012b4b3c..7b7e5459b 100755 --- a/actions/actions +++ b/actions/actions @@ -9,6 +9,7 @@ import logging import os import sys import traceback +import types import typing import plinth.log @@ -166,10 +167,9 @@ def _assert_valid_type(arg_name, value, annotation): return - if not hasattr(annotation, '__origin__'): - raise TypeError('Unsupported annotation type') - - if annotation.__origin__ == typing.Union: + # 'int | str' or 'typing.Union[int, str]' + if (isinstance(annotation, types.UnionType) + or getattr(annotation, '__origin__', None) == typing.Union): for arg in annotation.__args__: try: _assert_valid_type(arg_name, value, arg) @@ -179,7 +179,8 @@ def _assert_valid_type(arg_name, value, annotation): raise TypeError(f'Expected one of unioned types for {arg_name}') - if annotation.__origin__ == list: + # 'list[int]' or 'typing.List[int]' + if getattr(annotation, '__origin__', None) == list: if not isinstance(value, list): raise TypeError(f'Expected type list for {arg_name}') @@ -189,7 +190,8 @@ def _assert_valid_type(arg_name, value, annotation): return - if annotation.__origin__ == dict: + # 'list[dict]' or 'typing.List[dict]' + if getattr(annotation, '__origin__', None) == dict: if not isinstance(value, dict): raise TypeError(f'Expected type dict for {arg_name}') diff --git a/doc/dev/tutorial/customizing.rst b/doc/dev/tutorial/customizing.rst index 46c2ad867..5b0aadd2c 100644 --- a/doc/dev/tutorial/customizing.rst +++ b/doc/dev/tutorial/customizing.rst @@ -160,7 +160,6 @@ for transmission daemon. We will do this by creating a file ``privileged.py``. import json import pathlib - from typing import Union from plinth import action_utils from plinth.actions import privileged @@ -175,7 +174,7 @@ for transmission daemon. We will do this by creating a file ``privileged.py``. @privileged - def merge_configuration(configuration: dict[str, Union[str, bool]]) -> None: + def merge_configuration(configuration: dict[str, str | bool]) -> None: """Merge given JSON configuration with existing configuration.""" current_configuration = _transmission_config.read_bytes() current_configuration = json.loads(current_configuration) diff --git a/plinth/module_loader.py b/plinth/module_loader.py index 4e4992851..cc2a561dd 100644 --- a/plinth/module_loader.py +++ b/plinth/module_loader.py @@ -8,7 +8,6 @@ import importlib import logging import pathlib import re -from typing import Optional import django @@ -142,8 +141,7 @@ def get_module_import_path(module_name: str) -> str: return import_path -def _read_module_import_paths_from_file( - file_path: pathlib.Path) -> Optional[str]: +def _read_module_import_paths_from_file(file_path: pathlib.Path) -> str | None: """Read a module's import path from a file.""" with file_path.open() as file_handle: for line in file_handle: diff --git a/plinth/modules/backups/privileged.py b/plinth/modules/backups/privileged.py index d43cc4f6e..3ebc65178 100644 --- a/plinth/modules/backups/privileged.py +++ b/plinth/modules/backups/privileged.py @@ -7,7 +7,6 @@ import pathlib import re import subprocess import tarfile -from typing import Optional, Union from plinth.actions import privileged from plinth.utils import Version @@ -22,8 +21,8 @@ class AlreadyMountedError(Exception): @privileged -def mount(mountpoint: str, remote_path: str, ssh_keyfile: Optional[str] = None, - password: Optional[str] = None, +def mount(mountpoint: str, remote_path: str, ssh_keyfile: str | None = None, + password: str | None = None, user_known_hosts_file: str = '/dev/null'): """Mount a remote ssh path via sshfs.""" try: @@ -110,7 +109,7 @@ def setup(path: str): def _init_repository(path: str, encryption: str, - encryption_passphrase: Optional[str] = None): + encryption_passphrase: str | None = None): """Initialize a local or remote borg repository.""" if encryption != 'none': if not encryption_passphrase: @@ -121,14 +120,13 @@ def _init_repository(path: str, encryption: str, @privileged -def init(path: str, encryption: str, - encryption_passphrase: Optional[str] = None): +def init(path: str, encryption: str, encryption_passphrase: str | None = None): """Initialize the borg repository.""" _init_repository(path, encryption, encryption_passphrase) @privileged -def info(path: str, encryption_passphrase: Optional[str] = None) -> dict: +def info(path: str, encryption_passphrase: str | None = None) -> dict: """Show repository information.""" process = _run(['borg', 'info', '--json', path], encryption_passphrase, stdout=subprocess.PIPE) @@ -136,7 +134,7 @@ def info(path: str, encryption_passphrase: Optional[str] = None) -> dict: @privileged -def list_repo(path: str, encryption_passphrase: Optional[str] = None) -> dict: +def list_repo(path: str, encryption_passphrase: str | None = None) -> dict: """List repository contents.""" process = _run(['borg', 'list', '--json', '--format="{comment}"', path], encryption_passphrase, stdout=subprocess.PIPE) @@ -150,8 +148,8 @@ def _get_borg_version(): @privileged -def create_archive(path: str, paths: list[str], comment: Optional[str] = None, - encryption_passphrase: Optional[str] = None): +def create_archive(path: str, paths: list[str], comment: str | None = None, + encryption_passphrase: str | None = None): """Create archive.""" existing_paths = filter(os.path.exists, paths) command = ['borg', 'create', '--json'] @@ -169,7 +167,7 @@ def create_archive(path: str, paths: list[str], comment: Optional[str] = None, @privileged -def delete_archive(path: str, encryption_passphrase: Optional[str] = None): +def delete_archive(path: str, encryption_passphrase: str | None = None): """Delete archive.""" _run(['borg', 'delete', path], encryption_passphrase) @@ -199,7 +197,7 @@ def _extract(archive_path, destination, encryption_passphrase, locations=None): @privileged -def export_tar(path: str, encryption_passphrase: Optional[str] = None): +def export_tar(path: str, encryption_passphrase: str | None = None): """Export archive contents as tar stream on stdout.""" _run(['borg', 'export-tar', path, '-', '--tar-filter=gzip'], encryption_passphrase) @@ -214,7 +212,7 @@ def _read_archive_file(archive, filepath, encryption_passphrase): @privileged def get_archive_apps(path: str, - encryption_passphrase: Optional[str] = None) -> list[str]: + encryption_passphrase: str | None = None) -> list[str]: """Get list of apps included in archive.""" manifest_folder = os.path.relpath(MANIFESTS_FOLDER, '/') borg_call = [ @@ -286,7 +284,7 @@ def get_exported_archive_apps(path: str) -> list[str]: @privileged def restore_archive(archive_path: str, destination: str, directories: list[str], files: list[str], - encryption_passphrase: Optional[str] = None): + encryption_passphrase: str | None = None): """Restore files from an archive.""" locations_all = directories + files locations_all = [ @@ -319,8 +317,7 @@ def _assert_app_id(app_id): @privileged -def dump_settings(app_id: str, settings: dict[str, Union[int, float, bool, - str]]): +def dump_settings(app_id: str, settings: dict[str, int | float | bool | str]): """Dump an app's settings to a JSON file.""" _assert_app_id(app_id) BACKUPS_DATA_PATH.mkdir(exist_ok=True) @@ -329,7 +326,7 @@ def dump_settings(app_id: str, settings: dict[str, Union[int, float, bool, @privileged -def load_settings(app_id: str) -> dict[str, Union[int, float, bool, str]]: +def load_settings(app_id: str) -> dict[str, int | float | bool | str]: """Load an app's settings from a JSON file.""" _assert_app_id(app_id) settings_path = BACKUPS_DATA_PATH / f'{app_id}-settings.json' @@ -339,7 +336,7 @@ def load_settings(app_id: str) -> dict[str, Union[int, float, bool, str]]: return {} -def _get_env(encryption_passphrase: Optional[str] = None): +def _get_env(encryption_passphrase: str | None = None): """Create encryption and ssh kwargs out of given arguments.""" env = dict(os.environ, BORG_RELOCATED_REPO_ACCESS_IS_OK='yes', LANG='C.UTF-8') diff --git a/plinth/modules/bepasty/privileged.py b/plinth/modules/bepasty/privileged.py index d0e37bcb8..c9ddd7b63 100644 --- a/plinth/modules/bepasty/privileged.py +++ b/plinth/modules/bepasty/privileged.py @@ -11,7 +11,6 @@ import secrets import shutil import string import subprocess -from typing import Optional import augeas @@ -124,7 +123,7 @@ def get_configuration() -> dict[str, object]: @privileged -def add_password(permissions: list[str], comment: Optional[str] = None): +def add_password(permissions: list[str], comment: str | None = None): """Generate a password with given permissions.""" conf = conf_file_read() permissions = _format_permissions(permissions) diff --git a/plinth/modules/config/privileged.py b/plinth/modules/config/privileged.py index ad5b7f58d..953b8d639 100644 --- a/plinth/modules/config/privileged.py +++ b/plinth/modules/config/privileged.py @@ -4,7 +4,6 @@ import os import pathlib import subprocess -from typing import Optional import augeas @@ -29,7 +28,7 @@ def set_hostname(hostname: str): @privileged -def set_domainname(domainname: Optional[str] = None): +def set_domainname(domainname: str | None = None): """Set system domainname in /etc/hosts.""" hostname = subprocess.check_output(['hostname']).decode().strip() hosts_path = pathlib.Path('/etc/hosts') diff --git a/plinth/modules/ejabberd/__init__.py b/plinth/modules/ejabberd/__init__.py index 0dbb12e54..3ee7771b0 100644 --- a/plinth/modules/ejabberd/__init__.py +++ b/plinth/modules/ejabberd/__init__.py @@ -3,7 +3,6 @@ import json import logging -from typing import Tuple from django.urls import reverse_lazy from django.utils.translation import gettext_lazy as _ @@ -226,7 +225,7 @@ def update_turn_configuration(config: TurnConfiguration, managed=True, privileged.configure_turn(json.loads(config.to_json()), managed) -def get_turn_configuration() -> Tuple[TurnConfiguration, bool]: +def get_turn_configuration() -> tuple[TurnConfiguration, bool]: """Get the latest STUN/TURN configuration.""" tc, managed = privileged.get_turn_config() return TurnConfiguration(tc['domain'], tc['uris'], diff --git a/plinth/modules/ejabberd/privileged.py b/plinth/modules/ejabberd/privileged.py index e4dd971ce..7dd6e6498 100644 --- a/plinth/modules/ejabberd/privileged.py +++ b/plinth/modules/ejabberd/privileged.py @@ -8,7 +8,7 @@ import shutil import socket import subprocess from pathlib import Path -from typing import Any, Optional, Tuple +from typing import Any from ruamel.yaml import YAML, scalarstring @@ -240,7 +240,7 @@ def set_domains(domains: list[str]): @privileged -def mam(command: str) -> Optional[bool]: +def mam(command: str) -> bool | None: """Enable, disable, or get status of Message Archive Management (MAM).""" if command not in ('enable', 'disable', 'status'): raise ValueError('Invalid command') @@ -308,7 +308,7 @@ def _generate_uris(services: list[dict]) -> list[str]: @privileged -def get_turn_config() -> Tuple[dict[str, Any], bool]: +def get_turn_config() -> tuple[dict[str, Any], bool]: """Get the latest STUN/TURN configuration in JSON format.""" with open(EJABBERD_CONFIG, 'r', encoding='utf-8') as file_handle: conf = yaml.load(file_handle) diff --git a/plinth/modules/email/dns.py b/plinth/modules/email/dns.py index fecd051b7..2540fc8f9 100644 --- a/plinth/modules/email/dns.py +++ b/plinth/modules/email/dns.py @@ -10,7 +10,6 @@ See: https://rspamd.com/doc/modules/dkim_signing.html """ from dataclasses import dataclass -from typing import Union @dataclass @@ -19,12 +18,12 @@ class Entry: # pylint: disable=too-many-instance-attributes type_: str value: str - domain: Union[str, None] = None + domain: str | None = None class_: str = 'IN' ttl: int = 60 priority: int = 10 - weight: Union[int, None] = None - port: Union[int, None] = None + weight: int | None = None + port: int | None = None def get_split_value(self): """If the record is TXT and value > 255, split it.""" diff --git a/plinth/modules/gitweb/privileged.py b/plinth/modules/gitweb/privileged.py index bcd2dc5ce..2248ec776 100644 --- a/plinth/modules/gitweb/privileged.py +++ b/plinth/modules/gitweb/privileged.py @@ -9,7 +9,7 @@ import re import shutil import subprocess import time -from typing import Any, Optional +from typing import Any from plinth import action_utils from plinth.actions import privileged @@ -347,7 +347,7 @@ def repo_info(name: str) -> dict[str, str]: @privileged -def create_repo(url: Optional[str] = None, name: Optional[str] = None, +def create_repo(url: str | None = None, name: str | None = None, description: str = '', owner: str = '', keep_ownership: bool = False, is_private: bool = False, skip_prepare: bool = False, prepare_only: bool = False): diff --git a/plinth/modules/i2p/privileged.py b/plinth/modules/i2p/privileged.py index bd3c50395..9220bd698 100644 --- a/plinth/modules/i2p/privileged.py +++ b/plinth/modules/i2p/privileged.py @@ -1,8 +1,6 @@ # SPDX-License-Identifier: AGPL-3.0-or-later """Configure I2P.""" -from typing import Optional - from plinth.actions import privileged from plinth.modules.i2p.helpers import RouterEditor, TunnelEditor @@ -19,8 +17,8 @@ def set_tunnel_property(name: str, property_: str, value: str): @privileged -def add_favorite(name: str, url: str, description: Optional[str], - icon: Optional[str]): +def add_favorite(name: str, url: str, description: str | None, + icon: str | None): """Add a favorite to router.config.""" editor = RouterEditor() editor.read_conf().add_favorite(name, url, description, icon).write_conf() diff --git a/plinth/modules/ikiwiki/privileged.py b/plinth/modules/ikiwiki/privileged.py index ffd711889..5f6c6357f 100644 --- a/plinth/modules/ikiwiki/privileged.py +++ b/plinth/modules/ikiwiki/privileged.py @@ -6,7 +6,6 @@ import pathlib import re import shutil import subprocess -from typing import Tuple from plinth.actions import privileged @@ -43,7 +42,7 @@ def _get_title(site): @privileged -def get_sites() -> list[Tuple[str, str]]: +def get_sites() -> list[tuple[str, str]]: """Get wikis and blogs.""" sites = [] if os.path.exists(SITE_PATH): diff --git a/plinth/modules/matrixsynapse/privileged.py b/plinth/modules/matrixsynapse/privileged.py index f5b48459d..8f29caaec 100644 --- a/plinth/modules/matrixsynapse/privileged.py +++ b/plinth/modules/matrixsynapse/privileged.py @@ -7,7 +7,6 @@ import json import os import pathlib import shutil -from typing import Dict, List, Optional, Union import requests import yaml @@ -102,7 +101,7 @@ def get_config(): @privileged def set_config(public_registration: bool, - registration_verification: Optional[str] = None): + registration_verification: str | None = None): """Enable/disable public user registration.""" if registration_verification == 'token': _create_registration_token() @@ -243,7 +242,7 @@ def _get_access_token() -> str: @privileged -def list_registration_tokens() -> List[Dict[str, Optional[Union[str, int]]]]: +def list_registration_tokens() -> list[dict[str, str | int | None]]: """Return the current list of registration tokens.""" if not action_utils.service_is_running('matrix-synapse'): return [] @@ -262,7 +261,7 @@ def _get_headers(access_token: str): def _list_registration_tokens( - access_token: str) -> List[Dict[str, Optional[Union[str, int]]]]: + access_token: str) -> list[dict[str, str | int | None]]: """Use Admin API to fetch the list of registration tokens. For details on registration tokens API, see: diff --git a/plinth/modules/minetest/privileged.py b/plinth/modules/minetest/privileged.py index ceae1812f..73452d7b9 100644 --- a/plinth/modules/minetest/privileged.py +++ b/plinth/modules/minetest/privileged.py @@ -1,8 +1,6 @@ # SPDX-License-Identifier: AGPL-3.0-or-later """Configure Minetest server.""" -from typing import Optional - import augeas from plinth import action_utils @@ -13,10 +11,9 @@ AUG_PATH = '/files' + CONFIG_FILE + '/.anon' @privileged -def configure(max_players: Optional[int] = None, - enable_pvp: Optional[bool] = None, - creative_mode: Optional[bool] = None, - enable_damage: Optional[bool] = None): +def configure(max_players: int | None = None, enable_pvp: bool | None = None, + creative_mode: bool | None = None, + enable_damage: bool | None = None): """Update configuration file and restart daemon if necessary.""" aug = load_augeas() if max_players is not None: diff --git a/plinth/modules/mumble/privileged.py b/plinth/modules/mumble/privileged.py index 9ef198fab..9385a0f8c 100644 --- a/plinth/modules/mumble/privileged.py +++ b/plinth/modules/mumble/privileged.py @@ -5,7 +5,6 @@ Configure Mumble server. import pathlib import subprocess -from typing import Optional import augeas @@ -33,7 +32,7 @@ def set_super_user_password(password: str): @privileged -def get_domain() -> Optional[str]: +def get_domain() -> str | None: """Return domain name set in mumble or empty string.""" domain_file = pathlib.Path('/var/lib/mumble-server/domain-freedombox') try: @@ -43,7 +42,7 @@ def get_domain() -> Optional[str]: @privileged -def set_domain(domain_name: Optional[str]): +def set_domain(domain_name: str | None): """Write a file containing domain name.""" if domain_name: domain_file = pathlib.Path('/var/lib/mumble-server/domain-freedombox') @@ -69,7 +68,7 @@ def change_root_channel_name(root_channel_name: str): @privileged -def get_root_channel_name() -> Optional[str]: +def get_root_channel_name() -> str | None: """Return the currently configured Root channel name.""" aug = _load_augeas() name = aug.get('.anon/registerName') diff --git a/plinth/modules/privacy/privileged.py b/plinth/modules/privacy/privileged.py index 04faff907..0a5782258 100644 --- a/plinth/modules/privacy/privileged.py +++ b/plinth/modules/privacy/privileged.py @@ -2,7 +2,6 @@ """Configure Privacy App.""" import pathlib -from typing import Optional import augeas @@ -30,7 +29,7 @@ def setup(): @privileged -def set_configuration(enable_popcon: Optional[bool] = None): +def set_configuration(enable_popcon: bool | None = None): """Update popcon configuration.""" aug = _load_augeas() if enable_popcon: diff --git a/plinth/modules/shadowsocks/privileged.py b/plinth/modules/shadowsocks/privileged.py index 4d3075654..1776cc6d9 100644 --- a/plinth/modules/shadowsocks/privileged.py +++ b/plinth/modules/shadowsocks/privileged.py @@ -7,7 +7,6 @@ import pathlib import random import string from shutil import move -from typing import Union from plinth import action_utils from plinth.actions import privileged @@ -64,7 +63,7 @@ def setup(): @privileged -def get_config() -> dict[str, Union[int, str]]: +def get_config() -> dict[str, int | str]: """Read and print Shadowsocks configuration.""" config = open(SHADOWSOCKS_CONFIG_SYMLINK, 'r', encoding='utf-8').read() return json.loads(config) @@ -86,7 +85,7 @@ def _merge_config(config): @privileged -def merge_config(config: dict[str, Union[int, str]]): +def merge_config(config: dict[str, int | str]): """Configure Shadowsocks Client.""" _merge_config(config) diff --git a/plinth/modules/shadowsocksserver/privileged.py b/plinth/modules/shadowsocksserver/privileged.py index 1e51bdccd..1ea5f16d2 100644 --- a/plinth/modules/shadowsocksserver/privileged.py +++ b/plinth/modules/shadowsocksserver/privileged.py @@ -6,7 +6,6 @@ import os import pathlib import random import string -from typing import Union from plinth import action_utils from plinth.actions import privileged @@ -47,7 +46,7 @@ def setup(): @privileged -def get_config() -> dict[str, Union[int, str]]: +def get_config() -> dict[str, int | str]: """Read and print Shadowsocks Server configuration.""" config = open(SHADOWSOCKS_CONFIG_SYMLINK, 'r', encoding='utf-8').read() return json.loads(config) diff --git a/plinth/modules/tor/privileged.py b/plinth/modules/tor/privileged.py index 4e6b37f10..c32a31dc5 100644 --- a/plinth/modules/tor/privileged.py +++ b/plinth/modules/tor/privileged.py @@ -10,7 +10,7 @@ import shutil import socket import subprocess import time -from typing import Any, Optional, Union +from typing import Any import augeas @@ -143,11 +143,10 @@ def _remove_proxy(): @privileged -def configure(use_upstream_bridges: Optional[bool] = None, - upstream_bridges: Optional[str] = None, - relay: Optional[bool] = None, - bridge_relay: Optional[bool] = None, - hidden_service: Optional[bool] = None): +def configure(use_upstream_bridges: bool | None = None, + upstream_bridges: str | None = None, relay: bool | None = None, + bridge_relay: bool | None = None, + hidden_service: bool | None = None): """Configure Tor.""" aug = augeas_load() @@ -193,7 +192,7 @@ def restart(): @privileged -def get_status() -> dict[str, Union[bool, str, dict[str, Any]]]: +def get_status() -> dict[str, bool | str | dict[str, Any]]: """Return dict with Tor status.""" aug = augeas_load() return { @@ -323,8 +322,7 @@ def _disable(): action_utils.service_disable(SERVICE_NAME) -def _use_upstream_bridges(use_upstream_bridges: Optional[bool] = None, - aug=None): +def _use_upstream_bridges(use_upstream_bridges: bool | None = None, aug=None): """Enable use of upstream bridges.""" if use_upstream_bridges is None: return @@ -363,8 +361,7 @@ def _set_upstream_bridges(upstream_bridges=None, aug=None): aug.save() -def _enable_relay(relay: Optional[bool], bridge: Optional[bool], - aug: augeas.Augeas): +def _enable_relay(relay: bool | None, bridge: bool | None, aug: augeas.Augeas): """Enable Tor bridge relay.""" if relay is None and bridge is None: return diff --git a/plinth/modules/torproxy/privileged.py b/plinth/modules/torproxy/privileged.py index 57216af9d..46b12393d 100644 --- a/plinth/modules/torproxy/privileged.py +++ b/plinth/modules/torproxy/privileged.py @@ -5,7 +5,7 @@ import logging import os import shutil import subprocess -from typing import Any, Optional, Union +from typing import Any import augeas @@ -66,9 +66,9 @@ def _first_time_setup(): @privileged -def configure(use_upstream_bridges: Optional[bool] = None, - upstream_bridges: Optional[str] = None, - apt_transport_tor: Optional[bool] = None): +def configure(use_upstream_bridges: bool | None = None, + upstream_bridges: str | None = None, + apt_transport_tor: bool | None = None): """Configure Tor.""" aug = augeas_load() @@ -92,7 +92,7 @@ def restart(): @privileged -def get_status() -> dict[str, Union[bool, str, dict[str, Any]]]: +def get_status() -> dict[str, bool | str | dict[str, Any]]: """Return dict with Tor Proxy status.""" aug = augeas_load() return { @@ -125,8 +125,7 @@ def _disable(): action_utils.service_disable(SERVICE_NAME) -def _use_upstream_bridges(use_upstream_bridges: Optional[bool] = None, - aug=None): +def _use_upstream_bridges(use_upstream_bridges: bool | None = None, aug=None): """Enable use of upstream bridges.""" if use_upstream_bridges is None: return diff --git a/plinth/modules/transmission/privileged.py b/plinth/modules/transmission/privileged.py index 6c5992bdd..fbadd312a 100644 --- a/plinth/modules/transmission/privileged.py +++ b/plinth/modules/transmission/privileged.py @@ -5,7 +5,6 @@ Configuration helper for Transmission daemon. import json import pathlib -from typing import Union from plinth import action_utils from plinth.actions import privileged @@ -20,7 +19,7 @@ def get_configuration() -> dict[str, str]: @privileged -def merge_configuration(configuration: dict[str, Union[str, bool]]): +def merge_configuration(configuration: dict[str, str | bool]): """Merge given JSON configuration with existing configuration.""" current_configuration_bytes = _transmission_config.read_bytes() current_configuration = json.loads(current_configuration_bytes) diff --git a/plinth/modules/ttrss/privileged.py b/plinth/modules/ttrss/privileged.py index 34778b596..a59aaf3ca 100644 --- a/plinth/modules/ttrss/privileged.py +++ b/plinth/modules/ttrss/privileged.py @@ -3,7 +3,6 @@ import os import subprocess -from typing import Optional import augeas @@ -19,15 +18,16 @@ DB_BACKUP_FILE = '/var/lib/plinth/backups-data/ttrss-database.sql' @privileged def pre_setup(): """Preseed debconf values before packages are installed.""" - action_utils.debconf_set_selections( - ['tt-rss tt-rss/database-type string pgsql', - 'tt-rss tt-rss/purge boolean true', - 'tt-rss tt-rss/dbconfig-remove boolean true', - 'tt-rss tt-rss/dbconfig-reinstall boolean true']) + action_utils.debconf_set_selections([ + 'tt-rss tt-rss/database-type string pgsql', + 'tt-rss tt-rss/purge boolean true', + 'tt-rss tt-rss/dbconfig-remove boolean true', + 'tt-rss tt-rss/dbconfig-reinstall boolean true' + ]) @privileged -def get_domain() -> Optional[str]: +def get_domain() -> str | None: """Get the domain set for Tiny Tiny RSS.""" aug = load_augeas() @@ -41,7 +41,7 @@ def get_domain() -> Optional[str]: @privileged -def set_domain(domain_name: Optional[str]): +def set_domain(domain_name: str | None): """Set the domain to be used by Tiny Tiny RSS.""" if not domain_name: return diff --git a/plinth/modules/upgrades/privileged.py b/plinth/modules/upgrades/privileged.py index 1369e61b5..1b734d2b3 100644 --- a/plinth/modules/upgrades/privileged.py +++ b/plinth/modules/upgrades/privileged.py @@ -7,7 +7,6 @@ import pathlib import re import subprocess import time -from typing import List, Tuple, Union from plinth.action_utils import (apt_hold, apt_hold_flag, apt_hold_freedombox, apt_unhold_freedombox, debconf_set_selections, @@ -81,7 +80,7 @@ Pin: release a=bullseye-backports Pin-Priority: 500 ''' -DIST_UPGRADE_OBSOLETE_PACKAGES: List[str] = [] +DIST_UPGRADE_OBSOLETE_PACKAGES: list[str] = [] DIST_UPGRADE_PACKAGES_WITH_PROMPTS = [ 'bind9', 'firewalld', 'janus', 'minetest-server', 'minidlna', @@ -90,7 +89,7 @@ DIST_UPGRADE_PACKAGES_WITH_PROMPTS = [ DIST_UPGRADE_PRE_INSTALL_PACKAGES = ['base-files'] -DIST_UPGRADE_PRE_DEBCONF_SELECTIONS: List[str] = [ +DIST_UPGRADE_PRE_DEBCONF_SELECTIONS: list[str] = [ # Tell grub-pc to continue without installing grub again. 'grub-pc grub-pc/install_devices_empty boolean true' ] @@ -317,7 +316,7 @@ def _is_sufficient_free_space() -> bool: return free_space >= DIST_UPGRADE_REQUIRED_FREE_SPACE -def _check_dist_upgrade(test_upgrade=False) -> Tuple[bool, str]: +def _check_dist_upgrade(test_upgrade=False) -> tuple[bool, str]: """Check if a distribution upgrade be performed. Check for new stable release, if updates are enabled, and if there is @@ -589,7 +588,7 @@ def _start_dist_upgrade_service(): @privileged -def start_dist_upgrade(test: bool = False) -> dict[str, Union[str, bool]]: +def start_dist_upgrade(test: bool = False) -> dict[str, str | bool]: """Start dist upgrade process. Check if a new stable release is available, and start dist-upgrade process diff --git a/plinth/modules/users/privileged.py b/plinth/modules/users/privileged.py index 7483474b1..a67284e84 100644 --- a/plinth/modules/users/privileged.py +++ b/plinth/modules/users/privileged.py @@ -6,7 +6,6 @@ import os import re import shutil import subprocess -from typing import Optional import augeas @@ -220,8 +219,8 @@ def _disconnect_samba_user(username): @privileged -def create_user(username: str, password: str, auth_user: Optional[str] = None, - auth_password: Optional[str] = None): +def create_user(username: str, password: str, auth_user: str | None = None, + auth_password: str | None = None): """Create an LDAP user, set password and flush cache.""" _validate_user(auth_user, auth_password) @@ -232,7 +231,7 @@ def create_user(username: str, password: str, auth_user: Optional[str] = None, @privileged -def remove_user(username: str, password: Optional[str] = None): +def remove_user(username: str, password: str | None = None): """Remove an LDAP user.""" groups = _get_user_groups(username) @@ -424,8 +423,8 @@ def _add_user_to_group(username, groupname): @privileged def add_user_to_group(username: str, groupname: str, - auth_user: Optional[str] = None, - auth_password: Optional[str] = None): + auth_user: str | None = None, + auth_password: str | None = None): """Add an LDAP user to an LDAP group.""" if groupname == 'admin': _validate_user(auth_user, auth_password) diff --git a/plinth/modules/zoph/privileged.py b/plinth/modules/zoph/privileged.py index 65f342281..e5f3b79b9 100644 --- a/plinth/modules/zoph/privileged.py +++ b/plinth/modules/zoph/privileged.py @@ -5,7 +5,6 @@ import configparser import os import re import subprocess -from typing import Optional from plinth import action_utils from plinth.actions import privileged @@ -65,8 +64,8 @@ def _get_db_name(): @privileged -def set_configuration(enable_osm: Optional[bool] = None, - admin_user: Optional[str] = None): +def set_configuration(enable_osm: bool | None = None, + admin_user: str | None = None): """Setup Zoph Apache configuration.""" _zoph_configure('interface.user.remote', 'true') @@ -90,7 +89,7 @@ def set_configuration(enable_osm: Optional[bool] = None, @privileged -def is_configured() -> Optional[bool]: +def is_configured() -> bool | None: """Return whether zoph app is configured.""" try: process = subprocess.run( diff --git a/plinth/operation.py b/plinth/operation.py index 1985c8e08..ba6591e69 100644 --- a/plinth/operation.py +++ b/plinth/operation.py @@ -4,7 +4,7 @@ import enum import logging import threading -from typing import Callable, Optional +from typing import Callable from . import app as app_module @@ -22,10 +22,10 @@ class Operation: COMPLETED: str = 'completed' def __init__(self, app_id: str, name: str, target: Callable, - args: Optional[list] = None, kwargs: Optional[dict] = None, + args: list | None = None, kwargs: dict | None = None, show_message: bool = True, show_notification: bool = False, - thread_data: Optional[dict] = None, - on_complete: Callable = None): + thread_data: dict | None = None, + on_complete: Callable | None = None): """Initialize to no operation.""" self.app_id = app_id self.name = name @@ -39,8 +39,8 @@ class Operation: self.state = Operation.State.WAITING self.return_value = None - self._message: Optional[str] = None - self.exception: Optional[Exception] = None + self._message: str | None = None + self.exception: Exception | None = None # Operation specific data self.thread_data: dict = thread_data or {} @@ -94,8 +94,8 @@ class Operation: thread = threading.current_thread() return thread._operation # type: ignore [attr-defined] - def on_update(self, message: Optional[str] = None, - exception: Optional[Exception] = None): + def on_update(self, message: str | None = None, + exception: Exception | None = None): """Call from within the thread to update the progress of operation.""" if message: self._message = message @@ -172,7 +172,7 @@ class OperationsManager: def __init__(self): """Initialize the object.""" self._operations: list[Operation] = [] - self._current_operation: Optional[Operation] = None + self._current_operation: Operation | None = None # Assume that operations manager will be called from various threads # including the callback called from the threads it creates. Ensure diff --git a/plinth/package.py b/plinth/package.py index dee7b9baf..c9afab8b2 100644 --- a/plinth/package.py +++ b/plinth/package.py @@ -5,7 +5,6 @@ import enum import logging import pathlib import time -from typing import Optional, Union import apt.cache from django.utils.translation import gettext as _ @@ -42,11 +41,11 @@ class Package(PackageExpression): self, name, optional: bool = False, - version: Optional[str] = None, # ">=1.0,<2.0" - distribution: Optional[str] = None, # Debian, Ubuntu - suite: Optional[str] = None, # stable, testing - codename: Optional[str] = None, # bullseye-backports - architecture: Optional[str] = None): # arm64 + version: str | None = None, # ">=1.0,<2.0" + distribution: str | None = None, # Debian, Ubuntu + suite: str | None = None, # stable, testing + codename: str | None = None, # bullseye-backports + architecture: str | None = None): # arm64 self.name = name self.optional = optional self.version = version @@ -108,10 +107,10 @@ class Packages(app_module.FollowerComponent): REMOVE = 'remove' # Remove the packages before installing the app def __init__(self, component_id: str, - packages: list[Union[str, PackageExpression]], + packages: list[str | PackageExpression], skip_recommends: bool = False, - conflicts: Optional[list[str]] = None, - conflicts_action: Optional[ConflictsAction] = None): + conflicts: list[str] | None = None, + conflicts_action: ConflictsAction | None = None): """Initialize a new packages component. 'component_id' should be a unique ID across all components of an app @@ -230,14 +229,14 @@ class Packages(app_module.FollowerComponent): return results - def find_conflicts(self) -> Optional[list[str]]: + def find_conflicts(self) -> list[str] | None: """Return list of conflicting packages installed on the system.""" if not self.conflicts: return None return packages_installed(self.conflicts) - def has_unavailable_packages(self) -> Optional[bool]: + def has_unavailable_packages(self) -> bool | None: """Return whether any of the packages are not available. Returns True if one or more of the packages is not available in the @@ -465,7 +464,7 @@ def filter_conffile_prompt_packages(packages): return privileged.filter_conffile_packages(list(packages)) -def packages_installed(candidates: Union[list, tuple]) -> list: +def packages_installed(candidates: list | tuple) -> list: """Check which candidates are installed on the system. :param candidates: A list of package names. diff --git a/plinth/privileged/packages.py b/plinth/privileged/packages.py index a315f9144..9045d2fad 100644 --- a/plinth/privileged/packages.py +++ b/plinth/privileged/packages.py @@ -5,7 +5,7 @@ import logging import os import subprocess from collections import defaultdict -from typing import Any, Optional +from typing import Any import apt.cache import apt_inst @@ -31,7 +31,7 @@ def update(): @privileged def install(app_id: str, packages: list[str], skip_recommends: bool = False, - force_configuration: Optional[str] = None, reinstall: bool = False, + force_configuration: str | None = None, reinstall: bool = False, force_missing_configuration: bool = False): """Install packages using apt-get.""" if force_configuration not in ('old', 'new', None): diff --git a/plinth/tests/test_actions_actions.py b/plinth/tests/test_actions_actions.py index 05c18b20c..088fa6fa4 100644 --- a/plinth/tests/test_actions_actions.py +++ b/plinth/tests/test_actions_actions.py @@ -147,9 +147,9 @@ def test_assert_valid_type(actions_module): # Invalid values for int, str, float and Optional values = [[1, bool], ['foo', int], [1, str], [1, float], - [1, typing.Optional[str]], [1.1, typing.Union[int, str]], - [1, list], [1, dict], [[1], list[str]], - [{ + [1, typing.Optional[str]], [1, str | None], + [1.1, typing.Union[int, str]], [1.1, int | str], [1, list], + [1, dict], [[1], list[str]], [{ 'a': 'b' }, dict[int, str]], [{ 1: 2 @@ -165,8 +165,12 @@ def test_assert_valid_type(actions_module): assert_valid('arg', 1.1, float) assert_valid('arg', None, typing.Optional[int]) assert_valid('arg', 1, typing.Optional[int]) + assert_valid('arg', None, int | None) + assert_valid('arg', 1, int | None) assert_valid('arg', 1, typing.Union[int, str]) assert_valid('arg', '1', typing.Union[int, str]) + assert_valid('arg', 1, int | str) + assert_valid('arg', '1', int | str) assert_valid('arg', [], list[int]) assert_valid('arg', ['foo'], list[str]) assert_valid('arg', {}, dict[int, str])