improvement: replacing the use of LegacyVersion

This commit is contained in:
Vinicius Moreira
2022-11-25 06:57:22 -03:00
parent 69cf81dbe9
commit 93d2d66b71
15 changed files with 130 additions and 119 deletions

View File

@@ -1,21 +1,35 @@
import re
from typing import Tuple
from packaging.version import LegacyVersion
RE_VERSION_WITH_RELEASE = re.compile(r'^(.+)-\d+$')
RE_VERSION_WITH_EPOCH = re.compile(r'^\d+:(.+)$')
RE_VERSION_WITH_RELEASE = re.compile(r'^(.+)-\d+$')
def normalize_version(version: str) -> LegacyVersion:
final_version = version.strip()
def map_str_version(version: str) -> Tuple[str, ...]:
return tuple(part.zfill(8) for part in version.split("."))
if not RE_VERSION_WITH_EPOCH.match(final_version):
final_version = f'0:{final_version}'
if not RE_VERSION_WITH_RELEASE.match(final_version):
final_version = f'{final_version}-1'
def normalize_version(version: str) -> Tuple[int, Tuple[str, ...], int]:
raw_version = version.strip()
return LegacyVersion(final_version)
epoch = 0
if RE_VERSION_WITH_EPOCH.match(raw_version):
epoch_version = raw_version.split(":", maxsplit=1)
epoch = int(epoch_version[0])
raw_version = epoch_version[1]
release = 1
if RE_VERSION_WITH_RELEASE.match(raw_version):
version_release = raw_version.rsplit("-", maxsplit=1)
raw_version = version_release[0]
release = int(version_release[1])
if not raw_version.split(".")[0].isdigit():
# this is required to properly compare versions starting with a number against versions starting with alpha
raw_version = f"0.{raw_version}"
return epoch, map_str_version(raw_version), release
def match_required_version(current_version: str, operator: str, required_version: str) -> bool:
@@ -39,7 +53,7 @@ def match_required_version(current_version: str, operator: str, required_version
elif current_has_release and not required_has_release:
final_current = current_no_release[1]
final_required, final_current = LegacyVersion(final_required), LegacyVersion(final_current)
final_required, final_current = map_str_version(final_required), map_str_version(final_current)
if operator == '==' or operator == '=':
return final_current == final_required

View File

@@ -11,8 +11,6 @@ from pathlib import Path
from typing import Set, Type, List, Tuple, Optional, Iterable, Generator
from colorama import Fore
from packaging.version import LegacyVersion
from packaging.version import parse as parse_version
from bauh.api.abstract.context import ApplicationContext
from bauh.api.abstract.controller import SoftwareManager, SearchResult, UpgradeRequirements, UpgradeRequirement, \
@@ -28,6 +26,7 @@ from bauh.commons import resource
from bauh.commons.boot import CreateConfigFile
from bauh.commons.html import bold
from bauh.commons.system import SystemProcess, new_subprocess, ProcessHandler, SimpleProcess
from bauh.commons.version_util import normalize_version
from bauh.gems.appimage import query, INSTALLATION_DIR, APPIMAGE_SHARED_DIR, ROOT_DIR, \
APPIMAGE_CONFIG_DIR, UPDATES_IGNORED_FILE, util, get_default_manual_installation_file_dir, DATABASE_APPS_FILE, \
DATABASE_RELEASES_FILE, APPIMAGE_CACHE_DIR, get_icon_path, DOWNLOAD_DIR
@@ -286,11 +285,14 @@ class AppImageManager(SoftwareManager, SettingsController):
elif continuous_update and not continuous_version:
app.update = False
else:
try:
app.update = parse_version(tup[2]) > parse_version(app.version) if tup[2] else False
except:
app.update = False
traceback.print_exc()
if tup[2]:
try:
latest_version = normalize_version(tup[2])
installed_version = normalize_version(app.version)
app.update = latest_version > installed_version
except:
app.update = False
traceback.print_exc()
if app.update:
app.latest_version = tup[2]
@@ -509,14 +511,13 @@ class AppImageManager(SoftwareManager, SettingsController):
releases = cursor.execute(query.FIND_RELEASES_BY_APP_ID.format(app_tuple[0]))
if releases:
treated_releases = [(LegacyVersion(r[0]), *r[1:]) for r in releases]
treated_releases = [(normalize_version(r[0]), r[0], *r[1:]) for r in releases]
treated_releases.sort(key=self._sort_release, reverse=True)
for idx, tup in enumerate(treated_releases):
ver = str(tup[0])
history.append({'0_version': ver,
'1_published_at': datetime.strptime(tup[2], '%Y-%m-%dT%H:%M:%SZ') if tup[
2] else '', '2_url_download': tup[1]})
ver = tup[1]
date_ = datetime.strptime(tup[3], '%Y-%m-%dT%H:%M:%SZ') if tup[3] else ''
history.append({'0_version': ver, '1_published_at': date_, '2_url_download': tup[2]})
if res.pkg_status_idx == -1 and pkg.version == ver:
res.pkg_status_idx = idx

View File

@@ -5,10 +5,10 @@ from threading import Thread
from typing import Set, List, Tuple, Dict, Iterable, Optional, Generator, Pattern
from bauh.api.abstract.handler import ProcessWatcher
from bauh.commons.version_util import match_required_version
from bauh.gems.arch import pacman, message, sorting, confirmation
from bauh.gems.arch.aur import AURClient
from bauh.gems.arch.exceptions import PackageNotFoundException
from bauh.gems.arch.version import match_required_version
from bauh.view.util.translation import I18n

View File

@@ -5,7 +5,7 @@ from typing import Optional, List, Dict
from bauh.api.abstract.model import PackageStatus
from bauh.api.http import HttpClient
from bauh.gems.arch.model import ArchPackage
from bauh.gems.arch.version import normalize_version
from bauh.commons.version_util import normalize_version
from bauh.view.util.translation import I18n
URL_PKG_DOWNLOAD = 'https://aur.archlinux.org/{}'

View File

@@ -12,7 +12,7 @@ from bauh.gems.arch.dependencies import DependenciesAnalyser
from bauh.gems.arch.exceptions import PackageNotFoundException
from bauh.gems.arch.model import ArchPackage
from bauh.gems.arch.pacman import RE_DEP_OPERATORS
from bauh.gems.arch.version import match_required_version
from bauh.commons.version_util import match_required_version
from bauh.view.util.translation import I18n
@@ -22,8 +22,9 @@ class UpdateRequirementsContext:
aur_to_update: Dict[str, ArchPackage], repo_to_install: Dict[str, ArchPackage],
aur_to_install: Dict[str, ArchPackage], to_install: Dict[str, ArchPackage],
pkgs_data: Dict[str, dict], cannot_upgrade: Dict[str, UpgradeRequirement],
to_remove: Dict[str, UpgradeRequirement], installed_names: Dict[str, str], provided_map: Dict[str, Set[str]],
aur_index: Set[str], arch_config: dict, remote_provided_map: Dict[str, Set[str]], remote_repo_map: Dict[str, str],
to_remove: Dict[str, UpgradeRequirement], installed_names: Dict[str, str],
provided_map: Dict[str, Set[str]], aur_index: Set[str], arch_config: dict,
remote_provided_map: Dict[str, Set[str]], remote_repo_map: Dict[str, str],
root_password: Optional[str], aur_supported: bool):
self.to_update = to_update
self.repo_to_update = repo_to_update

View File

@@ -1,23 +1,21 @@
import os
from pathlib import Path
from packaging.version import parse as parse_version
from bauh import __app_name__
from bauh.api import user
from bauh.api.paths import CONFIG_DIR
from bauh.commons import resource
from bauh.commons.version_util import map_str_version
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_FILE = f'{CONFIG_DIR}/flatpak.yml'
FLATPAK_CONFIG_DIR = f'{CONFIG_DIR}/flatpak'
UPDATES_IGNORED_FILE = f'{FLATPAK_CONFIG_DIR}/updates_ignored.txt'
EXPORTS_PATH = '/usr/share/flatpak/exports/share' if user.is_root() else f'{Path.home()}/.local/share/flatpak/exports/share'
VERSION_1_2 = parse_version('1.2')
VERSION_1_3 = parse_version('1.3')
VERSION_1_4 = parse_version('1.4')
VERSION_1_5 = parse_version('1.5')
VERSION_1_12 = parse_version('1.12')
VERSION_1_2 = map_str_version("1.2")
VERSION_1_3 = map_str_version("1.3")
VERSION_1_4 = map_str_version("1.4")
VERSION_1_5 = map_str_version("1.5")
VERSION_1_12 = map_str_version("1.12")
def get_icon_path() -> str:

View File

@@ -7,8 +7,6 @@ from pathlib import Path
from threading import Thread
from typing import List, Set, Type, Tuple, Optional, Generator, Dict
from packaging.version import Version
from bauh.api import user
from bauh.api.abstract.controller import SearchResult, SoftwareManager, ApplicationContext, UpgradeRequirements, \
UpgradeRequirement, TransactionResult, SoftwareAction, SettingsView, SettingsController
@@ -123,7 +121,7 @@ class FlatpakManager(SoftwareManager, SettingsController):
res.total = len(res.installed) + len(res.new)
return res
def _add_updates(self, version: Version, output: list):
def _add_updates(self, version: Tuple[str, ...], output: list):
output.append(flatpak.list_updates_as_str(version))
def _fill_required_runtimes(self, installation: str, output: List[Tuple[str, str]]):
@@ -603,7 +601,7 @@ class FlatpakManager(SoftwareManager, SettingsController):
if file:
return file.text
def _fill_suggestion(self, appid: str, priority: SuggestionPriority, flatpak_version: Version, remote: str,
def _fill_suggestion(self, appid: str, priority: SuggestionPriority, flatpak_version: Tuple[str, ...], remote: str,
output: List[PackageSuggestion]):
app_json = flatpak.search(flatpak_version, appid, remote, app_id=True)

View File

@@ -6,13 +6,11 @@ from datetime import datetime
from threading import Thread
from typing import List, Dict, Set, Iterable, Optional, Tuple
from packaging.version import Version
from packaging.version import parse as parse_version
from bauh.api.exception import NoInternetException
from bauh.commons import system
from bauh.commons.system import new_subprocess, run_cmd, SimpleProcess, ProcessHandler, DEFAULT_LANG
from bauh.commons.util import size_to_byte
from bauh.commons.version_util import map_str_version
from bauh.gems.flatpak import EXPORTS_PATH, VERSION_1_3, VERSION_1_2, VERSION_1_5, VERSION_1_12
from bauh.gems.flatpak.constants import FLATHUB_URL
@@ -73,9 +71,9 @@ def is_installed():
return False if version is None else True
def get_version() -> Optional[Version]:
def get_version() -> Optional[Tuple[str, ...]]:
res = run_cmd('flatpak --version', print_error=False)
return parse_version(res.split(' ')[1].strip()) if res else None
return map_str_version(res.split(' ')[1].strip()) if res else None
def get_app_info(app_id: str, branch: str, installation: str) -> Optional[str]:
@@ -95,7 +93,7 @@ def get_commit(app_id: str, branch: str, installation: str) -> Optional[str]:
return commits[0][1].strip()
def list_installed(version: Version) -> List[dict]:
def list_installed(version: Tuple[str, ...]) -> List[dict]:
apps = []
if version < VERSION_1_2:
@@ -162,7 +160,8 @@ def list_installed(version: Version) -> List[dict]:
return apps
def update(app_ref: str, installation: str, version: Version, related: bool = False, deps: bool = False) -> SimpleProcess:
def update(app_ref: str, installation: str, version: Tuple[str, ...], related: bool = False, deps: bool = False) \
-> SimpleProcess:
cmd = ['flatpak', 'update', '-y', app_ref, f'--{installation}']
if not related:
@@ -180,7 +179,7 @@ def full_update(version: VERSION_1_12) -> SimpleProcess:
lang=DEFAULT_LANG if version < VERSION_1_12 else None)
def uninstall(app_ref: str, installation: str, version: Version) -> SimpleProcess:
def uninstall(app_ref: str, installation: str, version: Tuple[str, ...]) -> SimpleProcess:
return SimpleProcess(cmd=('flatpak', 'uninstall', app_ref, '-y', f'--{installation}'),
extra_paths={EXPORTS_PATH},
lang=DEFAULT_LANG if version < VERSION_1_12 else None,
@@ -191,7 +190,7 @@ def _new_updates() -> Dict[str, Set[str]]:
return {'full': set(), 'partial': set()}
def list_updates_as_str(version: Version) -> Dict[str, Set[str]]:
def list_updates_as_str(version: Tuple[str, ...]) -> Dict[str, Set[str]]:
sys_updates, user_updates = _new_updates(), _new_updates()
threads = []
@@ -226,7 +225,7 @@ def list_required_runtime_updates(installation: str) -> Optional[List[Tuple[str,
return RE_REQUIRED_RUNTIME.findall(updates)
def fill_updates(version: Version, installation: str, res: Dict[str, Set[str]]):
def fill_updates(version: Tuple[str, ...], installation: str, res: Dict[str, Set[str]]):
if version < VERSION_1_2:
try:
output = run_cmd(f'flatpak update --no-related --no-deps --{installation}', ignore_return_code=True)
@@ -274,7 +273,7 @@ def fill_updates(version: Version, installation: str, res: Dict[str, Set[str]]):
traceback.print_exc()
def downgrade(app_ref: str, commit: str, installation: str, root_password: Optional[str], version: Version) -> SimpleProcess:
def downgrade(app_ref: str, commit: str, installation: str, root_password: Optional[str], version: Tuple[str, ...]) -> SimpleProcess:
cmd = ('flatpak', 'update', '--no-related', '--no-deps', f'--commit={commit}', app_ref, '-y', f'--{installation}')
return SimpleProcess(cmd=cmd,
@@ -326,7 +325,7 @@ def get_app_commits_data(app_ref: str, origin: str, installation: str, full_str:
return commits
def search(version: Version, word: str, installation: str, app_id: bool = False) -> Optional[List[dict]]:
def search(version: Tuple[str, ...], word: str, installation: str, app_id: bool = False) -> Optional[List[dict]]:
res = run_cmd(f'flatpak search {word} --{installation}', lang=None)
@@ -408,7 +407,7 @@ def search(version: Version, word: str, installation: str, app_id: bool = False)
return found
def install(app_id: str, origin: str, installation: str, version: Version) -> SimpleProcess:
def install(app_id: str, origin: str, installation: str, version: Tuple[str, ...]) -> SimpleProcess:
return SimpleProcess(cmd=('flatpak', 'install', origin, app_id, '-y', f'--{installation}'),
extra_paths={EXPORTS_PATH},
lang=DEFAULT_LANG if version < VERSION_1_12 else None,
@@ -449,7 +448,7 @@ def run(app_id: str):
subprocess.Popen((f'flatpak run {app_id}',), shell=True, env={**os.environ})
def map_update_download_size(app_ids: Iterable[str], installation: str, version: Version) -> Dict[str, float]:
def map_update_download_size(app_ids: Iterable[str], installation: str, version: Tuple[str, ...]) -> Dict[str, float]:
success, output = ProcessHandler().handle_simple(SimpleProcess(('flatpak', 'update', f'--{installation}',
'--no-deps')))
if version >= VERSION_1_2:

View File

@@ -1,4 +1,4 @@
from packaging.version import Version
from typing import Tuple
from bauh.api.abstract.model import SoftwarePackage, PackageStatus
from bauh.commons import resource
@@ -140,7 +140,7 @@ class FlatpakApplication(SoftwarePackage):
if not self.runtime:
return super(FlatpakApplication, self).get_disk_icon_path()
def get_update_id(self, flatpak_version: Version) -> str:
def get_update_id(self, flatpak_version: Tuple[str, ...]) -> str:
if flatpak_version >= VERSION_1_2:
return f'{self.id}/{self.branch}/{self.installation}/{self.origin}'
else:

View File

@@ -2,12 +2,11 @@ import logging
import os
from pathlib import Path
from packaging.version import parse as parse_version
from bauh import __app_name__, __version__
from bauh.api.http import HttpClient
from bauh.api.paths import CACHE_DIR
from bauh.commons.html import bold, link
from bauh.commons.version_util import normalize_version
from bauh.view.util.translation import I18n
@@ -37,7 +36,7 @@ def check_for_update(logger: logging.Logger, http_client: HttpClient, i18n: I18n
release_file = '{}/{}{}'.format(notifications_dir, '' if not tray else 'tray_', latest['tag_name'])
if os.path.exists(release_file):
logger.info("Release {} already notified".format(latest['tag_name']))
elif parse_version(latest['tag_name']) > parse_version(__version__):
elif normalize_version(latest['tag_name']) > normalize_version(__version__):
try:
Path(notifications_dir).mkdir(parents=True, exist_ok=True)
with open(release_file, 'w+') as f: