mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-09 00:34:16 +02:00
[arch] improvmenet -> cleaning up some code warnings
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from typing import Iterable, List
|
from typing import Iterable, List, Optional
|
||||||
|
|
||||||
from bauh.api.abstract.handler import ProcessWatcher
|
from bauh.api.abstract.handler import ProcessWatcher
|
||||||
|
|
||||||
@@ -7,7 +7,7 @@ from bauh.api.abstract.handler import ProcessWatcher
|
|||||||
class FileDownloader(ABC):
|
class FileDownloader(ABC):
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def download(self, file_url: str, watcher: ProcessWatcher, output_path: str, cwd: str, root_password: str = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
|
def download(self, file_url: str, watcher: Optional[ProcessWatcher], output_path: str, cwd: str, root_password: str = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
|
||||||
"""
|
"""
|
||||||
:param file_url:
|
:param file_url:
|
||||||
:param watcher:
|
:param watcher:
|
||||||
|
|||||||
@@ -233,7 +233,7 @@ class AURClient:
|
|||||||
def clean_caches(self):
|
def clean_caches(self):
|
||||||
self.srcinfo_cache.clear()
|
self.srcinfo_cache.clear()
|
||||||
|
|
||||||
def map_update_data(self, pkgname: str, latest_version: str, srcinfo: dict = None) -> dict:
|
def map_update_data(self, pkgname: str, latest_version: Optional[str], srcinfo: Optional[dict] = None) -> dict:
|
||||||
info = self.get_src_info(pkgname) if not srcinfo else srcinfo
|
info = self.get_src_info(pkgname) if not srcinfo else srcinfo
|
||||||
|
|
||||||
provided = set()
|
provided = set()
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
from typing import Set, List, Tuple, Dict
|
from typing import Set, List, Tuple, Dict, Optional
|
||||||
|
|
||||||
from bauh.api.abstract.handler import ProcessWatcher
|
from bauh.api.abstract.handler import ProcessWatcher
|
||||||
from bauh.api.abstract.view import MultipleSelectComponent, InputOption, FormComponent, SingleSelectComponent, \
|
from bauh.api.abstract.view import MultipleSelectComponent, InputOption, FormComponent, SingleSelectComponent, \
|
||||||
@@ -44,7 +44,7 @@ def request_optional_deps(pkgname: str, pkg_repos: dict, watcher: ProcessWatcher
|
|||||||
return {o.value for o in view_opts.values}
|
return {o.value for o in view_opts.values}
|
||||||
|
|
||||||
|
|
||||||
def request_install_missing_deps(pkgname: str, deps: List[Tuple[str, str]], watcher: ProcessWatcher, i18n: I18n) -> bool:
|
def request_install_missing_deps(pkgname: Optional[str], deps: List[Tuple[str, str]], watcher: ProcessWatcher, i18n: I18n) -> bool:
|
||||||
msg = '<p>{}</p>'.format(i18n['arch.missing_deps.body'].format(name=bold(pkgname) if pkgname else '', deps=bold(str(len(deps)))))
|
msg = '<p>{}</p>'.format(i18n['arch.missing_deps.body'].format(name=bold(pkgname) if pkgname else '', deps=bold(str(len(deps)))))
|
||||||
|
|
||||||
opts = []
|
opts = []
|
||||||
@@ -101,4 +101,4 @@ def request_providers(providers_map: Dict[str, Set[str]], repo_map: Dict[str, st
|
|||||||
confirmation_label=i18n['proceed'].capitalize(),
|
confirmation_label=i18n['proceed'].capitalize(),
|
||||||
deny_label=i18n['cancel'].capitalize()):
|
deny_label=i18n['cancel'].capitalize()):
|
||||||
|
|
||||||
return {s.get_selected() for s in form.components}
|
return {s.get_selected() for s in form.components if isinstance(s, SingleSelectComponent)}
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ from datetime import datetime
|
|||||||
from math import floor
|
from math import floor
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from threading import Thread
|
from threading import Thread
|
||||||
from typing import List, Set, Type, Tuple, Dict, Iterable, Optional
|
from typing import List, Set, Type, Tuple, Dict, Iterable, Optional, Collection
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
from dateutil.parser import parse as parse_date
|
from dateutil.parser import parse as parse_date
|
||||||
@@ -927,7 +927,7 @@ class ArchManager(SoftwareManager):
|
|||||||
else:
|
else:
|
||||||
return [TextComponent(output)]
|
return [TextComponent(output)]
|
||||||
|
|
||||||
def list_related(self, pkgs: Iterable[str], all_pkgs: Iterable[str], data: Dict[str, dict], related: Set[str], provided_map: Dict[str, Set[str]]) -> Set[str]:
|
def list_related(self, pkgs: Collection[str], all_pkgs: Collection[str], data: Dict[str, dict], related: Set[str], provided_map: Dict[str, Set[str]]) -> Set[str]:
|
||||||
related.update(pkgs)
|
related.update(pkgs)
|
||||||
|
|
||||||
deps = set()
|
deps = set()
|
||||||
@@ -1261,7 +1261,7 @@ class ArchManager(SoftwareManager):
|
|||||||
|
|
||||||
return all_uninstalled
|
return all_uninstalled
|
||||||
|
|
||||||
def _request_uninstall_confirmation(self, to_uninstall: Iterable[str], required: Iterable[str], watcher: ProcessWatcher) -> bool:
|
def _request_uninstall_confirmation(self, to_uninstall: Collection[str], required: Collection[str], watcher: ProcessWatcher) -> bool:
|
||||||
reqs = [InputOption(label=p, value=p, icon_path=get_icon_path(), read_only=True) for p in required]
|
reqs = [InputOption(label=p, value=p, icon_path=get_icon_path(), read_only=True) for p in required]
|
||||||
reqs_select = MultipleSelectComponent(options=reqs, default_options=set(reqs), label="", max_per_line=1 if len(reqs) < 4 else 3)
|
reqs_select = MultipleSelectComponent(options=reqs, default_options=set(reqs), label="", max_per_line=1 if len(reqs) < 4 else 3)
|
||||||
|
|
||||||
@@ -1291,7 +1291,7 @@ class ArchManager(SoftwareManager):
|
|||||||
window_cancel=False):
|
window_cancel=False):
|
||||||
return {*reqs_select.get_selected_values()}
|
return {*reqs_select.get_selected_values()}
|
||||||
|
|
||||||
def _request_all_unncessary_uninstall_confirmation(self, pkgs: Iterable[str], context: TransactionContext):
|
def _request_all_unncessary_uninstall_confirmation(self, pkgs: Collection[str], context: TransactionContext):
|
||||||
reqs = [InputOption(label=p, value=p, icon_path=get_icon_path(), read_only=True) for p in pkgs]
|
reqs = [InputOption(label=p, value=p, icon_path=get_icon_path(), read_only=True) for p in pkgs]
|
||||||
reqs_select = MultipleSelectComponent(options=reqs, default_options=set(reqs), label="", max_per_line=1)
|
reqs_select = MultipleSelectComponent(options=reqs, default_options=set(reqs), label="", max_per_line=1)
|
||||||
|
|
||||||
@@ -1738,12 +1738,6 @@ class ArchManager(SoftwareManager):
|
|||||||
return res
|
return res
|
||||||
|
|
||||||
def _install_deps(self, context: TransactionContext, deps: List[Tuple[str, str]]) -> Iterable[str]:
|
def _install_deps(self, context: TransactionContext, deps: List[Tuple[str, str]]) -> Iterable[str]:
|
||||||
"""
|
|
||||||
:param pkgs_repos:
|
|
||||||
:param root_password:
|
|
||||||
:param handler:
|
|
||||||
:return: not installed dependency
|
|
||||||
"""
|
|
||||||
progress_increment = int(100 / len(deps))
|
progress_increment = int(100 / len(deps))
|
||||||
progress = 0
|
progress = 0
|
||||||
self._update_progress(context, 1)
|
self._update_progress(context, 1)
|
||||||
@@ -1820,7 +1814,7 @@ class ArchManager(SoftwareManager):
|
|||||||
|
|
||||||
self._update_progress(context, 100)
|
self._update_progress(context, 100)
|
||||||
|
|
||||||
def _map_repos(self, pkgnames: Iterable[str]) -> dict:
|
def _map_repos(self, pkgnames: Collection[str]) -> dict:
|
||||||
pkg_repos = pacman.get_repositories(pkgnames) # getting repositories set
|
pkg_repos = pacman.get_repositories(pkgnames) # getting repositories set
|
||||||
|
|
||||||
if len(pkgnames) != len(pkg_repos): # checking if any dep not found in the distro repos are from AUR
|
if len(pkgnames) != len(pkg_repos): # checking if any dep not found in the distro repos are from AUR
|
||||||
|
|||||||
@@ -138,8 +138,7 @@ class DependenciesAnalyser:
|
|||||||
return missing
|
return missing
|
||||||
return missing
|
return missing
|
||||||
|
|
||||||
def map_known_missing_deps(self, known_deps: Dict[str, str], watcher: ProcessWatcher, check_subdeps: bool = True) -> \
|
def map_known_missing_deps(self, known_deps: Dict[str, str], watcher: ProcessWatcher, check_subdeps: bool = True) -> Optional[List[Tuple[str, str]]]:
|
||||||
List[Tuple[str, str]]:
|
|
||||||
sorted_deps = [] # it will hold the proper order to install the missing dependencies
|
sorted_deps = [] # it will hold the proper order to install the missing dependencies
|
||||||
|
|
||||||
repo_deps, aur_deps = set(), set()
|
repo_deps, aur_deps = set(), set()
|
||||||
@@ -411,7 +410,7 @@ class DependenciesAnalyser:
|
|||||||
provided_map: Dict[str, Set[str]], remote_repo_map: Dict[str, str],
|
provided_map: Dict[str, Set[str]], remote_repo_map: Dict[str, str],
|
||||||
already_checked: Set[str], remote_provided_map: Dict[str, Set[str]],
|
already_checked: Set[str], remote_provided_map: Dict[str, Set[str]],
|
||||||
deps_data: Dict[str, dict], aur_idx: Iterable[str], sort: bool,
|
deps_data: Dict[str, dict], aur_idx: Iterable[str], sort: bool,
|
||||||
watcher: ProcessWatcher, automatch_providers: bool) -> List[Tuple[str, str]]:
|
watcher: ProcessWatcher, automatch_providers: bool) -> Optional[List[Tuple[str, str]]]:
|
||||||
"""
|
"""
|
||||||
:param missing_deps:
|
:param missing_deps:
|
||||||
:param provided_map:
|
:param provided_map:
|
||||||
|
|||||||
@@ -12,13 +12,13 @@ from bauh.commons.system import run_cmd, new_subprocess, new_root_subprocess, Sy
|
|||||||
from bauh.commons.util import size_to_byte
|
from bauh.commons.util import size_to_byte
|
||||||
from bauh.gems.arch.exceptions import PackageNotFoundException, PackageInHoldException
|
from bauh.gems.arch.exceptions import PackageNotFoundException, PackageInHoldException
|
||||||
|
|
||||||
RE_DEPS = re.compile(r'[\w\-_]+:[\s\w_\-\.]+\s+\[\w+\]')
|
RE_DEPS = re.compile(r'[\w\-_]+:[\s\w_\-.]+\s+\[\w+]')
|
||||||
RE_OPTDEPS = re.compile(r'[\w\._\-]+\s*:')
|
RE_OPTDEPS = re.compile(r'[\w._\-]+\s*:')
|
||||||
RE_DEP_NOTFOUND = re.compile(r'error:.+\'(.+)\'')
|
RE_DEP_NOTFOUND = re.compile(r'error:.+\'(.+)\'')
|
||||||
RE_DEP_OPERATORS = re.compile(r'[<>=]')
|
RE_DEP_OPERATORS = re.compile(r'[<>=]')
|
||||||
RE_INSTALLED_FIELDS = re.compile(r'(Name|Description|Version|Install Date|Validated By)\s*:\s*(.+)')
|
RE_INSTALLED_FIELDS = re.compile(r'(Name|Description|Version|Install Date|Validated By)\s*:\s*(.+)')
|
||||||
RE_INSTALLED_SIZE = re.compile(r'Installed Size\s*:\s*([0-9,\.]+)\s(\w+)\n?', re.IGNORECASE)
|
RE_INSTALLED_SIZE = re.compile(r'Installed Size\s*:\s*([0-9,.]+)\s(\w+)\n?', re.IGNORECASE)
|
||||||
RE_DOWNLOAD_SIZE = re.compile(r'Download Size\s*:\s*([0-9,\.]+)\s(\w+)\n?', re.IGNORECASE)
|
RE_DOWNLOAD_SIZE = re.compile(r'Download Size\s*:\s*([0-9,.]+)\s(\w+)\n?', re.IGNORECASE)
|
||||||
RE_UPDATE_REQUIRED_FIELDS = re.compile(r'(\bProvides\b|\bInstalled Size\b|\bConflicts With\b)\s*:\s(.+)\n')
|
RE_UPDATE_REQUIRED_FIELDS = re.compile(r'(\bProvides\b|\bInstalled Size\b|\bConflicts With\b)\s*:\s(.+)\n')
|
||||||
RE_REMOVE_TRANSITIVE_DEPS = re.compile(r'removing\s([\w\-_]+)\s.+required\sby\s([\w\-_]+)\n?')
|
RE_REMOVE_TRANSITIVE_DEPS = re.compile(r'removing\s([\w\-_]+)\s.+required\sby\s([\w\-_]+)\n?')
|
||||||
RE_AVAILABLE_MIRRORS = re.compile(r'.+\s+OK\s+.+\s+(\d+:\d+)\s+.+(http.+)')
|
RE_AVAILABLE_MIRRORS = re.compile(r'.+\s+OK\s+.+\s+(\d+:\d+)\s+.+(http.+)')
|
||||||
@@ -251,7 +251,7 @@ def check_missing(names: Set[str]) -> Set[str]:
|
|||||||
return not_installed
|
return not_installed
|
||||||
|
|
||||||
|
|
||||||
def read_repository_from_info(name: str) -> str:
|
def read_repository_from_info(name: str) -> Optional[str]:
|
||||||
info = new_subprocess(['pacman', '-Si', name])
|
info = new_subprocess(['pacman', '-Si', name])
|
||||||
|
|
||||||
not_found = False
|
not_found = False
|
||||||
@@ -459,7 +459,7 @@ def get_databases() -> Set[str]:
|
|||||||
with open('/etc/pacman.conf') as f:
|
with open('/etc/pacman.conf') as f:
|
||||||
conf_str = f.read()
|
conf_str = f.read()
|
||||||
|
|
||||||
return {db for db in re.findall(r'[\n|\s]+\[(\w+)\]', conf_str) if db != 'options'}
|
return {db for db in re.findall(r'[\n|\s]+\[(\w+)]', conf_str) if db != 'options'}
|
||||||
|
|
||||||
|
|
||||||
def can_refresh_mirrors() -> bool:
|
def can_refresh_mirrors() -> bool:
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
from typing import Dict, Set, Iterable, Tuple, List
|
from typing import Dict, Set, Tuple, List, Collection
|
||||||
|
|
||||||
|
|
||||||
def __add_dep_to_sort(pkgname: str, pkgs_data: Dict[str, dict], sorted_names: dict, not_sorted: Set[str],
|
def __add_dep_to_sort(pkgname: str, pkgs_data: Dict[str, dict], sorted_names: dict, not_sorted: Set[str],
|
||||||
@@ -35,7 +35,7 @@ def __add_dep_to_sort(pkgname: str, pkgs_data: Dict[str, dict], sorted_names: di
|
|||||||
return idx
|
return idx
|
||||||
|
|
||||||
|
|
||||||
def sort(pkgs: Iterable[str], pkgs_data: Dict[str, dict], provided_map: Dict[str, Set[str]] = None) -> List[Tuple[str, str]]:
|
def sort(pkgs: Collection[str], pkgs_data: Dict[str, dict], provided_map: Dict[str, Set[str]] = None) -> List[Tuple[str, str]]:
|
||||||
sorted_list, sorted_names, not_sorted = [], set(), set()
|
sorted_list, sorted_names, not_sorted = [], set(), set()
|
||||||
provided = provided_map if provided_map else {}
|
provided = provided_map if provided_map else {}
|
||||||
|
|
||||||
|
|||||||
@@ -366,7 +366,7 @@ class UpdatesSummarizer:
|
|||||||
|
|
||||||
return requirement
|
return requirement
|
||||||
|
|
||||||
def summarize(self, pkgs: List[ArchPackage], root_password: str, arch_config: dict) -> UpgradeRequirements:
|
def summarize(self, pkgs: List[ArchPackage], root_password: str, arch_config: dict) -> Optional[UpgradeRequirements]:
|
||||||
res = UpgradeRequirements([], [], [], [])
|
res = UpgradeRequirements([], [], [], [])
|
||||||
|
|
||||||
remote_provided_map = pacman.map_provided(remote=True)
|
remote_provided_map = pacman.map_provided(remote=True)
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import re
|
import re
|
||||||
|
|
||||||
RE_STRIP_EPIC = re.compile(r'^\d+:')
|
RE_STRIP_EPIC = re.compile(r'^\d+:')
|
||||||
RE_STRIP_RELEASE = re.compile(r'-[\d\.?]+$')
|
RE_STRIP_RELEASE = re.compile(r'-[\d.?]+$')
|
||||||
|
|
||||||
|
|
||||||
def clean_version(version: str) -> str:
|
def clean_version(version: str) -> str:
|
||||||
|
|||||||
Reference in New Issue
Block a user