mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 04:44:15 +02:00
Refactoring and improvements (#12)
* Removing controller * Refactoring FlatpakManager * Retrieving only installed apps data from the API * Removing wrong password logs
This commit is contained in:
Vinícius Moreira
GitHub
@@ -1,25 +1,100 @@
|
||||
from datetime import datetime, timedelta
|
||||
from threading import Lock, Thread
|
||||
from typing import List
|
||||
|
||||
from fpakman.core.model import FlatpakManager
|
||||
import requests
|
||||
|
||||
from fpakman.core import flatpak
|
||||
|
||||
__FLATHUB_URL__ = 'https://flathub.org'
|
||||
__FLATHUB_API_URL__ = __FLATHUB_URL__ + '/api/v1'
|
||||
|
||||
|
||||
class FlatpakController:
|
||||
class FlatpakManager:
|
||||
|
||||
def __init__(self, model: FlatpakManager):
|
||||
self.model = model
|
||||
def __init__(self, cache_expire: int = 60 * 60):
|
||||
self.cache_apps = {}
|
||||
self.cache_expire = cache_expire
|
||||
self.http_session = requests.Session()
|
||||
self.lock_db_read = Lock()
|
||||
self.lock_read = Lock()
|
||||
|
||||
def refresh(self) -> List[dict]:
|
||||
return self.model.read_installed()
|
||||
# TODO remove if not necessary for future releases
|
||||
def load_full_database_async(self):
|
||||
Thread(target=self.load_full_database, daemon=True).start()
|
||||
|
||||
def update(self, app_ref: str):
|
||||
return self.model.update_app(app_ref)
|
||||
# TODO remove if not necessary for future releases
|
||||
def load_full_database(self):
|
||||
|
||||
def check_installed(self) -> bool:
|
||||
version = self.model.get_version()
|
||||
return False if version is None else version
|
||||
self.lock_db_read.acquire()
|
||||
|
||||
def get_version(self) -> str:
|
||||
return self.model.get_version()
|
||||
try:
|
||||
res = self.http_session.get(__FLATHUB_API_URL__ + '/apps')
|
||||
|
||||
def uninstall(self, app_ref: str):
|
||||
return self.model.remove_app(app_ref)
|
||||
if res.status_code == 200:
|
||||
for app in res.json():
|
||||
self.cache_apps[app['flatpakAppId']] = app
|
||||
finally:
|
||||
self.lock_db_read.release()
|
||||
|
||||
def _request_app_data(self, app_id: str):
|
||||
|
||||
try:
|
||||
res = self.http_session.get('{}/apps/{}'.format(__FLATHUB_API_URL__, app_id), timeout=60)
|
||||
|
||||
if res.status_code == 200:
|
||||
return res.json()
|
||||
else:
|
||||
print("Could not retrieve app data for id '{}'. Server response: {}".format(app_id, res.status_code))
|
||||
except:
|
||||
print("Could not retrieve app data for id '{}'. Timeout".format(app_id))
|
||||
return None
|
||||
|
||||
def read_installed(self) -> List[dict]:
|
||||
|
||||
self.lock_read.acquire()
|
||||
|
||||
try:
|
||||
installed = flatpak.list_installed()
|
||||
|
||||
if installed:
|
||||
installed.sort(key=lambda p: p['name'].lower())
|
||||
|
||||
available_updates = flatpak.list_updates_as_str()
|
||||
|
||||
for app in installed:
|
||||
|
||||
app_data = self.cache_apps.get(app['id'])
|
||||
|
||||
if (not app['runtime'] and not app_data) or (app_data and app_data['expires_at'] <= datetime.utcnow()): # if data is not cached or expired, tries to retrieve it
|
||||
app_data = self._request_app_data(app['id'])
|
||||
app_data['expires_at'] = datetime.utcnow() + timedelta(seconds=self.cache_expire)
|
||||
self.cache_apps[app['id']] = app_data
|
||||
|
||||
if not app_data:
|
||||
app['latest_version'] = None
|
||||
app['icon'] = None
|
||||
else:
|
||||
app['latest_version'] = app_data['currentReleaseVersion']
|
||||
app['icon'] = app_data['iconMobileUrl']
|
||||
|
||||
if app['icon'].startswith('/'):
|
||||
app['icon'] = __FLATHUB_URL__ + app['icon']
|
||||
|
||||
app['update'] = app['id'] in available_updates
|
||||
|
||||
return installed
|
||||
|
||||
finally:
|
||||
self.lock_read.release()
|
||||
|
||||
def downgrade_app(self, app: dict, root_password: str):
|
||||
|
||||
commits = flatpak.get_app_commits(app['ref'], app['origin'])
|
||||
commit_idx = commits.index(app['commit'])
|
||||
|
||||
# downgrade is not possible if the app current commit in the first one:
|
||||
if commit_idx == len(commits) - 1:
|
||||
return None
|
||||
|
||||
return flatpak.downgrade_and_stream(app['ref'], commits[commit_idx + 1], root_password)
|
||||
|
||||
@@ -66,6 +66,11 @@ def get_app_info_fields(app_id: str, branch: str, fields: List[str] = [], check_
|
||||
return data
|
||||
|
||||
|
||||
def is_installed():
|
||||
version = get_version()
|
||||
return False if version is None else True
|
||||
|
||||
|
||||
def get_version():
|
||||
res = system.run_cmd('{} --version'.format(BASE_CMD))
|
||||
return res.split(' ')[1].strip() if res else None
|
||||
@@ -86,10 +91,6 @@ def list_installed() -> List[dict]:
|
||||
return []
|
||||
|
||||
|
||||
def update(app_ref: str) -> bool:
|
||||
return bool(system.run_cmd('{} update -y {}'.format(BASE_CMD, app_ref)))
|
||||
|
||||
|
||||
def update_and_stream(app_ref: str):
|
||||
"""
|
||||
Updates the app reference and streams Flatpak output,
|
||||
|
||||
@@ -1,142 +0,0 @@
|
||||
from threading import Lock, Thread
|
||||
from typing import List
|
||||
|
||||
import requests
|
||||
|
||||
from fpakman.core import flatpak
|
||||
|
||||
__FLATHUB_URL__ = 'https://flathub.org'
|
||||
__FLATHUB_API_URL__ = __FLATHUB_URL__ + '/api/v1'
|
||||
|
||||
|
||||
class ImpossibleDowngradeException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class FlatpakManager:
|
||||
|
||||
def __init__(self):
|
||||
self.apps = []
|
||||
self.apps_db = {}
|
||||
self.http_session = requests.Session()
|
||||
self.lock_db_read = Lock()
|
||||
|
||||
def load_database_async(self):
|
||||
Thread(target=self.load_database, daemon=True).start()
|
||||
|
||||
def load_database(self):
|
||||
|
||||
self.lock_db_read.acquire()
|
||||
|
||||
try:
|
||||
res = self.http_session.get(__FLATHUB_API_URL__ + '/apps')
|
||||
|
||||
if res.status_code == 200:
|
||||
for app in res.json():
|
||||
self.apps_db[app['flatpakAppId']] = app
|
||||
finally:
|
||||
self.lock_db_read.release()
|
||||
|
||||
def get_version(self):
|
||||
return flatpak.get_version()
|
||||
|
||||
def read_installed(self) -> List[dict]:
|
||||
|
||||
installed = flatpak.list_installed()
|
||||
|
||||
if installed:
|
||||
installed.sort(key=lambda p: p['name'].lower())
|
||||
|
||||
available_updates = flatpak.list_updates_as_str()
|
||||
|
||||
for app in installed:
|
||||
|
||||
if not self.apps_db:
|
||||
self.load_database()
|
||||
|
||||
if self.apps_db:
|
||||
app_data = self.apps_db.get(app['id'], None)
|
||||
else:
|
||||
app_data = None
|
||||
|
||||
if not app_data:
|
||||
app['latest_version'] = None
|
||||
app['icon'] = None
|
||||
else:
|
||||
app['latest_version'] = app_data['currentReleaseVersion']
|
||||
app['icon'] = app_data['iconMobileUrl']
|
||||
|
||||
if app['icon'].startswith('/'):
|
||||
app['icon'] = __FLATHUB_URL__ + app['icon']
|
||||
|
||||
app['update'] = app['id'] in available_updates
|
||||
|
||||
self.apps = installed
|
||||
return [*self.apps]
|
||||
|
||||
def update_apps(self, refs: List[str]) -> List[dict]:
|
||||
|
||||
if self.apps:
|
||||
|
||||
for ref in refs:
|
||||
package_found = [app for app in self.apps if app['ref'] == ref]
|
||||
|
||||
if package_found:
|
||||
package_found = package_found[0]
|
||||
updated = flatpak.update(ref)
|
||||
|
||||
if updated:
|
||||
package_found['update'] = not updated
|
||||
|
||||
return [*self.apps]
|
||||
|
||||
return []
|
||||
|
||||
def _find_by_ref(self, ref: str):
|
||||
package_found = [app for app in self.apps if app['ref'] == ref]
|
||||
|
||||
if package_found:
|
||||
return package_found[0]
|
||||
|
||||
def update_app(self, ref: str):
|
||||
|
||||
"""
|
||||
:param ref:
|
||||
:return: the update command stream
|
||||
"""
|
||||
|
||||
if self.apps:
|
||||
|
||||
app_found = self._find_by_ref(ref)
|
||||
|
||||
if app_found:
|
||||
return flatpak.update_and_stream(ref)
|
||||
|
||||
def remove_app(self, ref: str):
|
||||
"""
|
||||
:param ref:
|
||||
:return: the uninstall command stream
|
||||
"""
|
||||
if self.apps:
|
||||
|
||||
app_found = self._find_by_ref(ref)
|
||||
|
||||
if app_found:
|
||||
return flatpak.uninstall_and_stream(ref)
|
||||
|
||||
def downgrade_app(self, ref: str, root_password: str):
|
||||
|
||||
if self.apps:
|
||||
|
||||
app_found = self._find_by_ref(ref)
|
||||
|
||||
if app_found:
|
||||
commits = flatpak.get_app_commits(app_found['ref'], app_found['origin'])
|
||||
|
||||
commit_idx = commits.index(app_found['commit'])
|
||||
|
||||
# downgrade is not possible if the app current commit in the first one:
|
||||
if commit_idx == len(commits) - 1:
|
||||
raise ImpossibleDowngradeException()
|
||||
|
||||
return flatpak.downgrade_and_stream(app_found['ref'], commits[commit_idx + 1], root_password)
|
||||
Reference in New Issue
Block a user