gyt/opendan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor the code to make it comply with PEP-8 standards:Convert all class definitions to CamelCase style.

Browse Source 
(issue 37)
This commit is contained in:
Liu Zhicong
2023-08-23 11:19:16 -07:00
parent 23963adc6e
commit 5454009e7b
21 changed files with 601 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/component/package_manager/__init__.py
+3 -3
View File
@@ -1,3 +1,3 @@
from .env import pkg_env_mgr,pkg_env
from .pkg import pkg_info,pkg_media_info
from .installer import pkg_install_task
from .env import PackageEnvManager,PackageEnv
from .pkg import PackageInfo,PackageMediaInfo
from .installer import PackageInstallTask
src/component/package_manager/env.py
+22 -22
View File
@@ -2,16 +2,16 @@
import logging
import toml
from .pkg import pkg_info,pkg_media_info
from .installer import pkg_installer
from .pkg import PackageInfo,PackageMediaInfo
from .installer import PackageInstaller
logger = logging.getLogger(__name__)
class pkg_env_mgr:
class PackageEnvManager:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(pkg_env_mgr, cls).__new__(cls)
cls._instance = super(PackageEnvManager, cls).__new__(cls)
return cls._instance
def __init__(self) -> None:
@@ -19,46 +19,46 @@ class pkg_env_mgr:
pass
def get_env(self,cfg_path:str) -> pkg_env:
def get_env(self,cfg_path:str) -> PackageEnv:
if cfg_path in self._pkg_envs:
return self._pkg_envs[cfg_path]
else:
pkg_env = pkg_env(cfg_path)
pkg_env = PackageEnv(cfg_path)
self._pkg_envs[cfg_path] = pkg_env
return pkg_env
def get_user_env(self) -> pkg_env:
def get_user_env(self) -> PackageEnv:
pass
def get_system_env(self) -> pkg_env:
def get_system_env(self) -> PackageEnv:
pass
class pkg_env:
class PackageEnv:
def __init__(self,cfg_path:str) -> None:
self.pkg_dir : str = ""
self.pkg_obj_dir : str = ""
self.is_strict : bool = True
self.parent_envs : list[pkg_env] = None
self.parent_envs : list[PackageEnv] = None
self.index_dbs = None
self.cfg_path = cfg_path
self._load_pkg_cfg(cfg_path)
pass
def load(self,pkg_name:str,search_parent=True) -> pkg_media_info:
def load(self,pkg_name:str,search_parent=True) -> PackageMediaInfo:
pkg_path = None
pkg_id,verion_str,cid = pkg_info.parse_pkg_name(pkg_name)
pkg_id,verion_str,cid = PackageInfo.parse_pkg_name(pkg_name)
if cid is None:
if verion_str is None:
channel = self.get_pkg_channel_from_version(verion_str)
channel:str = self.get_pkg_channel_from_version(verion_str)
if channel is None:
pkg_path = f"{self.pkg_dir}{pkg_id}"
else:
pkg_path = f"{self.pkg_dir}{pkg_id}#{channel}"
else:
channel = self.get_pkg_channel_from_version(verion_str)
the_version = self.get_exact_version_from_installed(verion_str)
channel:str = self.get_pkg_channel_from_version(verion_str)
the_version:str = self.get_exact_version_from_installed(verion_str)
if the_version is None:
logger.warn(f"load {pkg_name} failed: no match version from {verion_str}")
return None
@@ -69,7 +69,7 @@ class pkg_env:
else:
pkg_path = f"{self.pkg_obj_dir}.{pkg_id}/{cid}"
media_info = self.try_load_pkg_media_info(pkg_id,pkg_path)
media_info:PackageMediaInfo = self.try_load_pkg_media_info(pkg_path)
if media_info is None:
if search_parent:
for parent_env in self.parent_envs:
@@ -86,28 +86,28 @@ class pkg_env:
def get_pkg_channel_from_version(self,pkg_version:str) -> str:
pass
def get_pkg_media_info(self,pkg_name:str)->pkg_media_info:
def get_pkg_media_info(self,pkg_name:str)->PackageMediaInfo:
pass
def try_load_pkg_media_info(self,pkg_full_path:str) -> pkg_media_info:
def try_load_pkg_media_info(self,pkg_full_path:str) -> PackageMediaInfo:
pass
def get_installed_pkg_info(self,pkg_name:str) -> pkg_info:
def get_installed_pkg_info(self,pkg_name:str) -> PackageInfo:
pass
def lookup(self,pkg_id:str,version_str:str) -> pkg_info:
def lookup(self,pkg_id:str,version_str:str) -> PackageInfo:
# to make sure pkg.cid is correct, we MUST verfiy eveything here
pass
def get_installer(self) -> pkg_installer:
def get_installer(self) -> PackageInstaller:
pass
@classmethod
def is_valied_media(pkg_full_path:str) -> bool:
pass
def do_pkg_media_trans(self,pkg_info:pkg_info,source_path:str,target_path:str) -> bool:
def do_pkg_media_trans(self,pkg_info:PackageInfo,source_path:str,target_path:str) -> bool:
pass
def _load_pkg_cfg(self,cfg_path:str):
src/component/package_manager/installer.py
+16 -17
View File
@@ -5,10 +5,9 @@ import aiohttp
import aiofiles
import os
from ndn_client import content_id,ndn_client
from .pkg import pkg_info,pkg_media_info
from .env import pkg_env
from ndn_client import ContentId,NDN_Client
from .pkg import PackageInfo,PackageMediaInfo
from .env import PackageEnv
logger = logging.getLogger(__name__)
@@ -19,8 +18,8 @@ INSTALL_TASK_STATE_DOWNLOADING = 3
INSTALL_TASK_STATE_INSTALLING = 4
INSTALL_TAKS_STATE_ERROR = 5
class pkg_install_task:
def __init__(self,owner:pkg_env) -> None:
class PackageInstallTask:
def __init__(self,owner:PackageEnv) -> None:
self.owner = owner
self.state = INSTALL_TASK_STATE_CHECK_DEPENDENCY
@@ -29,20 +28,20 @@ class pkg_install_task:
self.dependency_tasks = None
self.error_str = None
class pkg_installer:
def __init__(self,owner_env:pkg_env) -> None:
class PackageInstaller:
def __init__(self,owner_env:PackageEnv) -> None:
self.all_tasks = {}
self.owner_env = owner_env
def install(self,pkg_name:str,
install_from_dependency = False, can_upgrade = True,skip_depends = False,options = None)->Tuple[pkg_install_task,str]:
install_from_dependency = False, can_upgrade = True,skip_depends = False,options = None)->Tuple[PackageInstallTask,str]:
the_pkg_info : pkg_info = None
the_pkg_info : PackageInfo = None
is_upgrade : bool = False
need_backup : bool = False
pkg_id,version_str,cid = pkg_info.parse_pkg_name(pkg_name)
media_info : pkg_media_info = self.owner_env.get_media_info(pkg_name) # must use index-db?
pkg_id,version_str,cid = PackageInfo.parse_pkg_name(pkg_name)
media_info : PackageMediaInfo = self.owner_env.get_media_info(pkg_name) # must use index-db?
if media_info is not None:
if cid is not None:
if can_upgrade:
@@ -74,7 +73,7 @@ class pkg_installer:
return result_task,"already installing"
logger.info(f"start download&install {pkg_name},install_from_dependency={install_from_dependency},upgrade={is_upgrade},backup={need_backup},target_pkg_info={the_pkg_info}")
result_task = pkg_install_task()
result_task = PackageInstallTask(self.owner_env)
self.all_tasks[the_pkg_info.cid] = result_task
async def download_and_install_pkg()->int:
# check dependency
@@ -104,7 +103,7 @@ class pkg_installer:
urls = self.owner_env.get_pkg_urls(the_pkg_info)
#download
client = ndn_client() # set watch
client = NDN_Client() # set watch
download_result = await client.get_file(the_pkg_info.cid,urls,target_full_path,options)
if download_result !=0:
result_task.state = INSTALL_TAKS_STATE_ERROR
@@ -140,16 +139,16 @@ class pkg_installer:
def uninstall(self):
pass
def get_dependency_tasks(self,pkg:pkg_info,dependency_tasks):
def get_dependency_tasks(self,pkg:PackageInfo,dependency_tasks):
pass
async def check_dependency(self,pkg:pkg_info,task_list:{}) -> bool:
async def check_dependency(self,pkg:PackageInfo,task_list:{}) -> bool:
for depend_pkg_name in pkg.depends:
depend_task = task_list.get(depend_pkg_name)
if depend_task is not None:
logger.debug(f"{pkg.name}'s depend pkg {depend_pkg_name} already in task list")
continue
depend_task = pkg_install_task()
depend_task = PackageInstallTask(self.owner_env)
task_list[depend_pkg_name] = depend_task
depend_pkg_info = self.owner_env.lookup(depend_pkg_name)
src/component/package_manager/pkg.py
+4 -2
View File
@@ -1,5 +1,7 @@
from typing import Tuple
class pkg_info:
class PackageInfo:
def __init__(self) -> None:
self.name = ""
self.cid = None
@@ -20,7 +22,7 @@ class pkg_info:
def cid(self) -> str:
return self.cid
class pkg_media_info:
class PackageMediaInfo:
def __init__(self) -> None:
pass