archive/blender-package-manager-addon
Archived
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Initial multiple repository support

Browse Source 
And lots of code reshuffling which likely should've been done in
separate commits..
This commit is contained in:
Ellwood Zwovic
2017-08-12 19:36:39 -07:00
parent 7340d8dadb
commit 6691b0b226
8 changed files with 463 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

384
package_manager/__init__.py
View File

@@ -33,82 +33,27 @@ if 'bpy' in locals():
messages = recursive_reload(messages) messages = recursive_reload(messages)
utils = recursive_reload(utils) utils = recursive_reload(utils)
bpkg = recursive_reload(bpkg) bpkg = recursive_reload(bpkg)
Package = bpkg.Package Package = bpkg.types.Package
else: else:
from . import subproc from . import subproc
from . import messages from . import messages
from . import bpkg from . import bpkg
from . import utils from . import utils
from .bpkg import Package from .bpkg.types import (
Package,
ConsolidatedPackage,
)
import bpy import bpy
from pathlib import Path
from collections import OrderedDict from collections import OrderedDict
# global list of all known packages, indexed by name # global list of all known packages, indexed by name
_packages = OrderedDict() _packages = OrderedDict()
class ConsolidatedPackage: # used for lazy loading
""" _main_has_run = False
Stores a grouping of different versions of the same package
"""
log = logging.getLogger(__name__ + ".ConsolidatedPackage")
def __init__(self, pkg=None):
self.versions = []
self.updateable = False
if pkg is not None:
self.add_version(pkg)
@property
def installed(self) -> bool:
"""Return true if any version of this package is installed"""
for pkg in self.versions:
if pkg.installed:
return True
return False
@property
def name(self) -> str:
"""
Return name of this package. All package versions in a
ConsolidatedPackage should have the same name by definition
"""
return self.versions[0].name
def get_latest_installed_version(self) -> bpkg.Package:
"""
Return the installed package with the highest version number.
If no packages are installed, return None.
"""
#self.versions is always sorted newer -> older, so we can just grab the first we find
for pkg in self.versions:
if pkg.installed:
return pkg
return None
def get_latest_version(self) -> bpkg.Package:
"""Return package with highest version number"""
return self.versions[0] # this is always sorted with the highest on top
def get_display_version(self) -> bpkg.Package:
"""
Return installed package with highest version number.
If no version is installed, return highest uninstalled version.
"""
pkg = self.get_latest_installed_version()
if pkg is None:
pkg = self.get_latest_version()
return pkg
def add_version(self, pkg: Package):
self.versions.append(pkg)
self.versions.sort(key=lambda v: v.version, reverse=True)
def __iter__(self):
return (pkg for pkg in self.versions)
class SubprocMixin: class SubprocMixin:
"""Mix-in class for things that need to be run in a subprocess.""" """Mix-in class for things that need to be run in a subprocess."""
@@ -306,7 +251,8 @@ class PACKAGE_OT_install(SubprocMixin, bpy.types.Operator):
def _subproc_success(self, success: messages.Success): def _subproc_success(self, success: messages.Success):
self.report({'INFO'}, 'Package installed successfully') self.report({'INFO'}, 'Package installed successfully')
bpy.ops.package.refresh_packages() global _packages
_packages = build_packagelist()
self.quit() self.quit()
def _subproc_aborted(self, aborted: messages.Aborted): def _subproc_aborted(self, aborted: messages.Aborted):
@@ -379,56 +325,58 @@ class PACKAGE_OT_uninstall(SubprocMixin, bpy.types.Operator):
self.report({'WARNING'}, 'Error downloading package, but process finished OK. This is weird.') self.report({'WARNING'}, 'Error downloading package, but process finished OK. This is weird.')
def get_installed_packages(refresh=False) -> list:
def get_packages_from_disk(refresh=False) -> list:
"""Get list of packages installed on disk""" """Get list of packages installed on disk"""
import addon_utils import addon_utils
return [Package.from_module(mod) for mod in addon_utils.modules(refresh=refresh)] return [Package.from_module(mod) for mod in addon_utils.modules(refresh=refresh)]
def get_packages_from_repo() -> list: def get_repo_storage_path() -> Path:
"""Get list of packages from cached repository lists (does not refresh them from server)""" return Path(bpy.utils.user_resource('CONFIG', 'repositories'))
import pathlib
storage_path = pathlib.Path(bpy.utils.user_resource('CONFIG', 'packages', create=True)) / 'repo.json'
try:
repo = bpkg.Repository.from_file(storage_path)
except FileNotFoundError:
return []
for pkg in repo.packages:
pkg.repository = repo.name
return repo.packages
class PACKAGE_OT_refresh_packages(bpy.types.Operator): def get_repositories() -> list:
bl_idname = "package.refresh_packages" """
bl_label = "Refresh Packages" Get list of downloaded repositories and update wm.package_repositories
bl_description = "Scan for packages on disk" """
log = logging.getLogger(__name__ + ".get_repositories")
storage_path = get_repo_storage_path()
repos = bpkg.load_repositories(storage_path)
log.debug("repos: %s", repos)
log = logging.getLogger(__name__ + ".PACKAGE_OT_refresh_packages") return repos
def execute(self, context): # class PACKAGE_OT_refresh_packages(bpy.types.Operator):
global _packages # bl_idname = "package.refresh_packages"
installed_packages = get_packages_from_disk(refresh=True) # bl_label = "Refresh Packages"
available_packages = get_packages_from_repo() # bl_description = "Scan for packages on disk"
_packages = build_composite_packagelist(installed_packages, available_packages) #
context.area.tag_redraw() # log = logging.getLogger(__name__ + ".PACKAGE_OT_refresh_packages")
#
# def execute(self, context):
# global _packages
# installed_packages = get_packages_from_disk(refresh=True)
# available_packages = get_packages_from_repo()
# _packages = build_composite_packagelist(installed_packages, available_packages)
# context.area.tag_redraw()
#
# return {'FINISHED'}
return {'FINISHED'} class PACKAGE_OT_refresh(SubprocMixin, bpy.types.Operator):
bl_idname = "package.refresh"
class PACKAGE_OT_refresh_repositories(SubprocMixin, bpy.types.Operator): bl_label = "Refresh"
bl_idname = "package.refresh_repositories"
bl_label = "Refresh Repositories"
bl_description = 'Check repositories for new and updated packages' bl_description = 'Check repositories for new and updated packages'
bl_options = {'REGISTER'} bl_options = {'REGISTER'}
log = logging.getLogger(__name__ + ".PACKAGE_OT_refresh_repositories") log = logging.getLogger(__name__ + ".PACKAGE_OT_refresh")
_running = False _running = False
def invoke(self, context, event): def invoke(self, context, event):
self.repolist = bpy.context.user_preferences.addons[__package__].preferences.repositories wm = context.window_manager
if len(self.repolist) == 0: self.repositories = wm.package_repositories
if len(self.repositories) == 0:
self.report({'ERROR'}, "No repositories to refresh") self.report({'ERROR'}, "No repositories to refresh")
return {'CANCELLED'} return {'CANCELLED'}
PACKAGE_OT_refresh_repositories._running = True PACKAGE_OT_refresh._running = True
return super().invoke(context, event) return super().invoke(context, event)
@classmethod @classmethod
@@ -436,7 +384,7 @@ class PACKAGE_OT_refresh_repositories(SubprocMixin, bpy.types.Operator):
return not cls._running return not cls._running
def cancel(self, context): def cancel(self, context):
PACKAGE_OT_refresh_repositories._running = False PACKAGE_OT_refresh._running = False
context.area.tag_redraw() context.area.tag_redraw()
def create_subprocess(self): def create_subprocess(self):
@@ -455,18 +403,19 @@ class PACKAGE_OT_refresh_repositories(SubprocMixin, bpy.types.Operator):
messages.SubprocError: self._subproc_error, messages.SubprocError: self._subproc_error,
messages.DownloadError: self._subproc_download_error, messages.DownloadError: self._subproc_download_error,
messages.Success: self._subproc_success, messages.Success: self._subproc_success,
messages.RepositoryResult: self._subproc_repository_result, # messages.RepositoryResult: self._subproc_repository_result,
messages.BadRepositoryError: self._subproc_repository_error, messages.BadRepositoryError: self._subproc_repository_error,
messages.Aborted: self._subproc_aborted, messages.Aborted: self._subproc_aborted,
} }
import pathlib import pathlib
storage_path = pathlib.Path(bpy.utils.user_resource('CONFIG', 'packages', create=True)) storage_path = pathlib.Path(bpy.utils.user_resource('CONFIG', 'repositories', create=True))
repository_url = self.repolist[0].url repository_urls = [repo.url for repo in self.repositories]
self.log.debug("Repository urls %s", repository_urls)
proc = multiprocessing.Process(target=subproc.refresh_repository, proc = multiprocessing.Process(target=subproc.refresh_repositories,
args=(self.pipe_subproc, storage_path, repository_url)) args=(self.pipe_subproc, storage_path, repository_urls))
return proc return proc
def _subproc_progress(self, progress: messages.Progress): def _subproc_progress(self, progress: messages.Progress):
@@ -485,19 +434,19 @@ class PACKAGE_OT_refresh_repositories(SubprocMixin, bpy.types.Operator):
self.quit() self.quit()
def _subproc_success(self, success: messages.Success): def _subproc_success(self, success: messages.Success):
self.report({'INFO'}, 'Finished refreshing lists')
self.quit() self.quit()
def _subproc_repository_result(self, result: messages.RepositoryResult): # def _subproc_repository_result(self, result: messages.RepositoryResult):
available_packages = result.repository.packages # available_packages = result.repository.packages
installed_packages = get_packages_from_disk(refresh=False) # installed_packages = get_packages_from_disk(refresh=False)
#
# TODO: deduplicate creation of view-packages.. # # TODO: deduplicate creation of view-packages..
for pkg in available_packages: # for pkg in available_packages:
pkg.repository = result.repository.name # pkg.repository = result.repository.name
#
global _packages # global _packages
_packages = build_composite_packagelist(installed_packages, available_packages) # _packages = build_composite_packagelist(installed_packages, available_packages)
self.report({'INFO'}, 'Package list retrieved successfully')
def _subproc_aborted(self, aborted: messages.Aborted): def _subproc_aborted(self, aborted: messages.Aborted):
self.report({'ERROR'}, 'Package list retrieval aborted per your request') self.report({'ERROR'}, 'Package list retrieval aborted per your request')
@@ -511,26 +460,15 @@ class PACKAGE_OT_refresh_repositories(SubprocMixin, bpy.types.Operator):
self.log.error('Refresh process died without telling us! Exit code was 0 though') self.log.error('Refresh process died without telling us! Exit code was 0 though')
self.report({'WARNING'}, 'Error refreshing package lists, but process finished OK. This is weird.') self.report({'WARNING'}, 'Error refreshing package lists, but process finished OK. This is weird.')
#TODO:
# monkey patch refresh_repositories and add refresh_packages in the success callback
# this way refresh_packages is always called after repositories have been refreshed
class PACKAGE_OT_refresh(bpy.types.Operator):
bl_idname = "package.refresh"
bl_label = "Refresh"
bl_description = "Check for new and updated packages"
def execute(self, context):
bpy.ops.package.refresh_repositories()
# getattr(bpy.ops, __package__).refresh_packages()
return {'FINISHED'}
class RepositoryProperty(bpy.types.PropertyGroup): class RepositoryProperty(bpy.types.PropertyGroup):
name = bpy.props.StringProperty(name="Name")
url = bpy.props.StringProperty(name="URL") url = bpy.props.StringProperty(name="URL")
status = bpy.props.EnumProperty(name="Status", items=[ status = bpy.props.EnumProperty(name="Status", items=[
("OK", "Okay", "FILE_TICK"), ("OK", "Okay", "FILE_TICK"),
("NOTFOUND", "Not found", "ERROR"), ("NOTFOUND", "Not found", "ERROR"),
("NOCONNECT", "Could not connect", "QUESTION"), ("NOCONNECT", "Could not connect", "QUESTION"),
]) ])
enabled = bpy.props.BoolProperty(name="Enabled")
class PACKAGE_UL_repositories(bpy.types.UIList): class PACKAGE_UL_repositories(bpy.types.UIList):
def draw_item(self, context, layout, data, item, icon, active_data, active_propname): def draw_item(self, context, layout, data, item, icon, active_data, active_propname):
@@ -551,17 +489,16 @@ class PACKAGE_OT_add_repository(bpy.types.Operator):
return wm.invoke_props_dialog(self) return wm.invoke_props_dialog(self)
def execute(self, context): def execute(self, context):
prefs = context.user_preferences.addons[__package__].preferences wm = context.window_manager
if len(prefs.repositories) > 0:
self.report({'ERROR'}, "Only one repository at a time is currently supported")
return {'CANCELLED'}
if len(self.url) == 0: if len(self.url) == 0:
self.report({'ERROR'}, "Repository URL not specified") self.report({'ERROR'}, "Repository URL not specified")
return {'CANCELLED'} return {'CANCELLED'}
repo = prefs.repositories.add() repo = wm.package_repositories.add()
repo.url = utils.parse_repository_url(self.url) repo.url = utils.sanitize_repository_url(self.url)
bpy.ops.package.refresh()
context.area.tag_redraw() context.area.tag_redraw()
return {'FINISHED'} return {'FINISHED'}
@@ -571,8 +508,19 @@ class PACKAGE_OT_remove_repository(bpy.types.Operator):
bl_label = "Remove Repository" bl_label = "Remove Repository"
def execute(self, context): def execute(self, context):
prefs = context.user_preferences.addons[__package__].preferences wm = context.window_manager
prefs.repositories.remove(prefs.active_repository) try:
repo = wm['package_repositories'][wm.package_active_repository]
except IndexError:
return {'CANCELLED'}
filename = bpkg.utils.format_filename(repo.name) + ".json"
path = (get_repo_storage_path() / filename)
if path.exists():
path.unlink()
wm.package_repositories.remove(wm.package_active_repository)
return {'FINISHED'} return {'FINISHED'}
class USERPREF_PT_packages(bpy.types.Panel): class USERPREF_PT_packages(bpy.types.Panel):
@@ -583,11 +531,11 @@ class USERPREF_PT_packages(bpy.types.Panel):
log = logging.getLogger(__name__ + '.USERPREF_PT_packages') log = logging.getLogger(__name__ + '.USERPREF_PT_packages')
# available_packages = []
# installed_packages = []
displayed_packages = [] displayed_packages = []
expanded_packages = [] expanded_packages = []
redraw = True
@classmethod @classmethod
def poll(cls, context): def poll(cls, context):
userpref = context.user_preferences userpref = context.user_preferences
@@ -598,19 +546,19 @@ class USERPREF_PT_packages(bpy.types.Panel):
wm = context.window_manager wm = context.window_manager
prefs = context.user_preferences.addons[__package__].preferences prefs = context.user_preferences.addons[__package__].preferences
main = layout.row() mainrow = layout.row()
spl = main.split(.2) spl = mainrow.split(.2)
sidebar = spl.column(align=True) sidebar = spl.column(align=True)
pkgzone = spl.column() pkgzone = spl.column()
sidebar.label("Repositories") sidebar.label("Repositories")
row = sidebar.row() row = sidebar.row()
row.template_list("PACKAGE_UL_repositories", "", prefs, "repositories", prefs, "active_repository") row.template_list("PACKAGE_UL_repositories", "", wm, "package_repositories", wm, "package_active_repository")
col = row.column(align=True) col = row.column(align=True)
col.operator(PACKAGE_OT_add_repository.bl_idname, text="", icon='ZOOMIN') col.operator(PACKAGE_OT_add_repository.bl_idname, text="", icon='ZOOMIN')
col.operator(PACKAGE_OT_remove_repository.bl_idname, text="", icon='ZOOMOUT') col.operator(PACKAGE_OT_remove_repository.bl_idname, text="", icon='ZOOMOUT')
sidebar.separator() sidebar.separator()
sidebar.operator(PACKAGE_OT_refresh_repositories.bl_idname) sidebar.operator(PACKAGE_OT_refresh.bl_idname, text="Check for updates")
sidebar.separator() sidebar.separator()
sidebar.label("Category") sidebar.label("Category")
@@ -865,15 +813,13 @@ class USERPREF_PT_packages(bpy.types.Panel):
row.alignment='CENTER' row.alignment='CENTER'
row.scale_y = 10 row.scale_y = 10
global _packages global _main_has_run
if not _main_has_run:
if len(_packages) == 0:
# TODO: read repository and installed packages synchronously for now; # TODO: read repository and installed packages synchronously for now;
# can't run an operator from draw code to do async monitoring # can't run an operator from draw code to do async monitoring
installed_packages = get_packages_from_disk() main()
available_packages = get_packages_from_repo()
_packages = build_composite_packagelist(installed_packages, available_packages) global _packages
if len(_packages) == 0: if len(_packages) == 0:
center_message(pkgzone, "No packages found") center_message(pkgzone, "No packages found")
return return
@@ -892,7 +838,7 @@ class USERPREF_PT_packages(bpy.types.Panel):
draw_package(_packages[pkgname], row) draw_package(_packages[pkgname], row)
class WM_OT_package_toggle_expand(bpy.types.Operator): class WM_OT_package_toggle_expand(bpy.types.Operator):# {{{
bl_idname = "wm.package_toggle_expand" bl_idname = "wm.package_toggle_expand"
bl_label = "" bl_label = ""
bl_description = "Toggle display of extended information for given package (hold shift to collapse all other packages)" bl_description = "Toggle display of extended information for given package (hold shift to collapse all other packages)"
@@ -913,9 +859,9 @@ class WM_OT_package_toggle_expand(bpy.types.Operator):
else: else:
USERPREF_PT_packages.expanded_packages.append(self.package_name) USERPREF_PT_packages.expanded_packages.append(self.package_name)
return {'FINISHED'} return {'FINISHED'}# }}}
class PACKAGE_OT_toggle_enabled(bpy.types.Operator): class PACKAGE_OT_toggle_enabled(bpy.types.Operator):# {{{
bl_idname = "package.toggle_enabled" bl_idname = "package.toggle_enabled"
bl_label = "" bl_label = ""
bl_description = "Enable given package if it's disabled, and vice versa if it's enabled" bl_description = "Enable given package if it's disabled, and vice versa if it's enabled"
@@ -954,9 +900,9 @@ class PACKAGE_OT_toggle_enabled(bpy.types.Operator):
addon_utils.enable(pkg.module_name, default_set=True) addon_utils.enable(pkg.module_name, default_set=True)
pkg.enabled = True pkg.enabled = True
return {'FINISHED'} return {'FINISHED'}# }}}
class PACKAGE_OT_disable(bpy.types.Operator): class PACKAGE_OT_disable(bpy.types.Operator):# {{{
bl_idname = "package.disable" bl_idname = "package.disable"
bl_label = "" bl_label = ""
bl_description = "Disable given package" bl_description = "Disable given package"
@@ -979,96 +925,69 @@ class PACKAGE_OT_disable(bpy.types.Operator):
ret = bpy.ops.wm.addon_disable(package.module_name) ret = bpy.ops.wm.addon_disable(package.module_name)
if ret == {'FINISHED'}: if ret == {'FINISHED'}:
_packages[self.package_name].enabled = False _packages[self.package_name].enabled = False
return ret return ret# }}}
class PackageManagerPreferences(bpy.types.AddonPreferences): # class PackageManagerPreferences(bpy.types.AddonPreferences):
bl_idname = __package__ # bl_idname = __package__
#
# repositories = bpy.props.CollectionProperty(
# type=RepositoryProperty,
# name="Repositories",
# )
# active_repository = bpy.props.IntProperty()
repositories = bpy.props.CollectionProperty( def build_packagelist() -> OrderedDict:# {{{
type=RepositoryProperty, """Make an OrderedDict of ConsolidatedPackages from known repositories + installed packages, keyed by package name"""
name="Repositories",
)
active_repository = bpy.props.IntProperty()
def validate_packagelist(pkglist: list) -> list:
"""Ensures all packages have required fields; strips out bad packages and returns them in a list"""
pass
def build_composite_packagelist(installed: list, available: list) -> OrderedDict:
"""Merge list of installed and available packages into one dict, keyed by package name"""
log = logging.getLogger(__name__ + ".build_composite_packagelist") log = logging.getLogger(__name__ + ".build_composite_packagelist")
masterlist = {} masterlist = {}
installed_packages = get_installed_packages()
known_repositories = get_repositories()
def packages_are_equivilent(pkg1: Package, pkg2: Package) -> bool: for repo in known_repositories:
"""Check that packages are the same version and provide the same files""" for pkg in repo.packages:
return pkg1.version == pkg2.version\ if pkg.name is None:
and pkg1.files == pkg2.files return OrderedDict()
def is_user_package(pkg: Package) -> bool:
"""Check if package's install location is in user scripts path or preferences scripts path"""
from pathlib import Path
user_script_path = bpy.utils.script_path_user()
prefs_script_path = bpy.utils.script_path_pref()
if user_script_path is not None:
in_user = Path(user_script_path) in Path(pkg.installed_location).parents
else:
in_user = False
if prefs_script_path is not None:
in_prefs = Path(prefs_script_path) in Path(pkg.installed_location).parents
else:
in_prefs = False
return in_user or in_prefs
def is_enabled(pkg: Package) -> bool:
"""Check if package is enabled"""
if pkg.module_name is not None:
return (pkg.module_name in bpy.context.user_preferences.addons)
else:
raise ValueError("Can't determine if package is enabled without knowing its module name")
for pkg in available:
pkgname = pkg.bl_info['name']
if pkgname in masterlist:
masterlist[pkgname].add_version(pkg)
else:
masterlist[pkgname] = ConsolidatedPackage(pkg)
for pkg in installed:
pkg.installed = True
pkg.enabled = is_enabled(pkg)
pkg.user = is_user_package(pkg)
if pkg.name in masterlist: if pkg.name in masterlist:
for masterpkg in masterlist[pkg.name]:
if packages_are_equivilent(pkg, masterpkg):
masterpkg.installed = True
masterpkg.installed_location = pkg.installed_location
masterpkg.user = pkg.user
masterpkg.module_name = pkg.module_name
masterpkg.enabled = pkg.enabled
break
else:
if masterpkg.version > pkg.version:
masterlist[pkg.name].updateable = True
masterlist[pkg.name].add_version(pkg) masterlist[pkg.name].add_version(pkg)
else: else:
masterlist[pkg.name] = ConsolidatedPackage(pkg) masterlist[pkg.name] = ConsolidatedPackage(pkg)
return OrderedDict(sorted(masterlist.items())) for pkg in installed_packages:
if pkg.name in masterlist:
masterlist[pkg.name].add_version(pkg)
else:
masterlist[pkg.name] = ConsolidatedPackage(pkg)
# log.debug(masterlist[None].__dict__)
return OrderedDict(sorted(masterlist.items()))# }}}
def main():
"""Entry point; performs initial loading of repositories and installed packages"""
global _packages
global _main_has_run
_packages = build_packagelist()
repos = get_repositories()
wm = bpy.context.window_manager
wm.package_repositories.clear()
for repo in repos:
repo_prop = wm.package_repositories.add()
repo_prop.name = repo.name
repo_prop.url = repo.url
# needed for lazy loading
_main_has_run = True
def register(): def register():
bpy.utils.register_class(PACKAGE_OT_install) bpy.utils.register_class(PACKAGE_OT_install)
bpy.utils.register_class(PACKAGE_OT_uninstall) bpy.utils.register_class(PACKAGE_OT_uninstall)
bpy.utils.register_class(PACKAGE_OT_toggle_enabled) bpy.utils.register_class(PACKAGE_OT_toggle_enabled)
# bpy.utils.register_class(PACKAGE_OT_disable) # bpy.utils.register_class(PACKAGE_OT_disable)
bpy.utils.register_class(PACKAGE_OT_refresh_repositories) # bpy.utils.register_class(PACKAGE_OT_refresh_repositories)
bpy.utils.register_class(PACKAGE_OT_refresh_packages) # bpy.utils.register_class(PACKAGE_OT_refresh_packages)
bpy.utils.register_class(PACKAGE_OT_refresh) bpy.utils.register_class(PACKAGE_OT_refresh)
bpy.utils.register_class(USERPREF_PT_packages) bpy.utils.register_class(USERPREF_PT_packages)
bpy.utils.register_class(WM_OT_package_toggle_expand) bpy.utils.register_class(WM_OT_package_toggle_expand)
@@ -1087,11 +1006,16 @@ def register():
) )
bpy.utils.register_class(RepositoryProperty) bpy.utils.register_class(RepositoryProperty)
bpy.types.WindowManager.package_repositories = bpy.props.CollectionProperty(
type=RepositoryProperty,
name="Repositories",
)
bpy.types.WindowManager.package_active_repository = bpy.props.IntProperty()
bpy.utils.register_class(PACKAGE_OT_add_repository) bpy.utils.register_class(PACKAGE_OT_add_repository)
bpy.utils.register_class(PACKAGE_OT_remove_repository) bpy.utils.register_class(PACKAGE_OT_remove_repository)
bpy.utils.register_class(PACKAGE_UL_repositories) bpy.utils.register_class(PACKAGE_UL_repositories)
bpy.utils.register_class(PackageManagerPreferences) # bpy.utils.register_class(PackageManagerPreferences)
def unregister(): def unregister():
@@ -1099,8 +1023,8 @@ def unregister():
bpy.utils.unregister_class(PACKAGE_OT_uninstall) bpy.utils.unregister_class(PACKAGE_OT_uninstall)
bpy.utils.unregister_class(PACKAGE_OT_toggle_enabled) bpy.utils.unregister_class(PACKAGE_OT_toggle_enabled)
# bpy.utils.unregister_class(PACKAGE_OT_disable) # bpy.utils.unregister_class(PACKAGE_OT_disable)
bpy.utils.unregister_class(PACKAGE_OT_refresh_repositories) # bpy.utils.unregister_class(PACKAGE_OT_refresh_repositories)
bpy.utils.unregister_class(PACKAGE_OT_refresh_packages) # bpy.utils.unregister_class(PACKAGE_OT_refresh_packages)
bpy.utils.unregister_class(PACKAGE_OT_refresh) bpy.utils.unregister_class(PACKAGE_OT_refresh)
bpy.utils.unregister_class(USERPREF_PT_packages) bpy.utils.unregister_class(USERPREF_PT_packages)
bpy.utils.unregister_class(WM_OT_package_toggle_expand) bpy.utils.unregister_class(WM_OT_package_toggle_expand)
@@ -1108,8 +1032,10 @@ def unregister():
del bpy.types.WindowManager.package_install_filter del bpy.types.WindowManager.package_install_filter
bpy.utils.unregister_class(RepositoryProperty) bpy.utils.unregister_class(RepositoryProperty)
del bpy.types.WindowManager.package_repositories
del bpy.types.WindowManager.package_active_repository
bpy.utils.unregister_class(PACKAGE_OT_add_repository) bpy.utils.unregister_class(PACKAGE_OT_add_repository)
bpy.utils.unregister_class(PACKAGE_OT_remove_repository) bpy.utils.unregister_class(PACKAGE_OT_remove_repository)
bpy.utils.unregister_class(PACKAGE_UL_repositories) bpy.utils.unregister_class(PACKAGE_UL_repositories)
bpy.utils.unregister_class(PackageManagerPreferences) # bpy.utils.unregister_class(PackageManagerPreferences)

27
package_manager/bpkg/__init__.py
View File

@@ -1,18 +1,19 @@
import logging __all__ = (
"exceptions",
"types",
)
log = logging.getLogger(__name__)
if 'bpy' in locals():
import importlib
log.debug("Reloading %s", __name__)
exceptions = importlib.reload(exceptions)
Package = importlib.reload(types.Package)
Repository = importlib.reload(types.Repository)
else:
from . import exceptions
from . types import ( from . types import (
Package, Package,
Repository, Repository,
) )
from pathlib import Path
def load_repositories(repo_storage_path: Path) -> list:
repositories = []
for repofile in repo_storage_path.glob('*.json'):
# try
repo = Repository.from_file(repofile)
# except
repositories.append(repo)
return repositories

3
package_manager/bpkg/exceptions.py
View File

@@ -9,3 +9,6 @@ class DownloadException(BpkgException):
class BadRepositoryException(BpkgException): class BadRepositoryException(BpkgException):
"""Raised when there is an error while reading or manipulating a repository""" """Raised when there is an error while reading or manipulating a repository"""
class PackageException(BpkgException):
"""Raised when there is an error while manipulating a package"""

271
package_manager/bpkg/types.py
View File

@@ -15,14 +15,57 @@ class Package:
self.bl_info = {} self.bl_info = {}
self.url = "" self.url = ""
self.files = [] self.files = []
self.set_from_dict(package_dict)
self.installed = False self.repositories = []
self.enabled = False
self.repository = None
self.installed_location = None self.installed_location = None
self.module_name = None self.module_name = None
self.set_from_dict(package_dict)
@property
def is_user(self) -> bool:
"""Return true if package's install location is in user or preferences scripts path"""
import bpy
user_script_path = bpy.utils.script_path_user()
prefs_script_path = bpy.utils.script_path_pref()
if user_script_path is not None:
in_user = Path(user_script_path) in Path(self.installed_location).parents
else:
in_user = False
if prefs_script_path is not None:
in_prefs = Path(prefs_script_path) in Path(self.installed_location).parents
else:
in_prefs = False
return in_user or in_prefs
@property
def enabled(self) -> bool:
"""Return true if package is enabled"""
import bpy
if self.module_name is not None:
return (self.module_name in bpy.context.user_preferences.addons)
else:
return False
@property
def installed(self) -> bool:
"""Return true if package is installed"""
import addon_utils
return len([Package.from_module(mod) for mod in addon_utils.modules(refresh=False) if
addon_utils.module_bl_info(mod)['name'] == self.name and
addon_utils.module_bl_info(mod)['version'] == self.version]) > 0
def set_installed_metadata(self, installed_pkg):
"""Sets metadata specific to installed packages from the Package given as `installed_pkg`"""
# self.installed = installed_pkg.installed
# self.enabled = installed_pkg.enabled
self.module_name = installed_pkg.module_name
self.installed_location = installed_pkg.installed_location
# self.is_user = installed_pkg.is_user
def to_dict(self) -> dict: def to_dict(self) -> dict:
""" """
Return a dict representation of the package Return a dict representation of the package
@@ -49,91 +92,58 @@ class Package:
@property @property
def name(self) -> str: def name(self) -> str:
"""Get name from bl_info""" """Get name from bl_info"""
try: return self.bl_info.get('name')
return self.bl_info['name']
except KeyError:
return None
@property @property
def version(self) -> tuple: def version(self) -> tuple:
"""Get version from bl_info""" """Get version from bl_info"""
try: return tuple(self.bl_info.get('version'))
return tuple(self.bl_info['version'])
except KeyError:
return None
@property @property
def blender(self) -> tuple: def blender(self) -> tuple:
"""Get blender from bl_info""" """Get blender from bl_info"""
try: return self.bl_info.get('blender')
return self.bl_info['blender']
except KeyError:
return None
# optional fields # optional fields
@property @property
def description(self) -> str: def description(self) -> str:
"""Get description from bl_info""" """Get description from bl_info"""
try: return self.bl_info.get('description')
return self.bl_info['description']
except KeyError:
return None
@property @property
def author(self) -> str: def author(self) -> str:
"""Get author from bl_info""" """Get author from bl_info"""
try: return self.bl_info.get('author')
return self.bl_info['author']
except KeyError:
return None
@property @property
def category(self) -> str: def category(self) -> str:
"""Get category from bl_info""" """Get category from bl_info"""
try: return self.bl_info.get('category')
return self.bl_info['category']
except KeyError:
return None
@property @property
def location(self) -> str: def location(self) -> str:
"""Get location from bl_info""" """Get location from bl_info"""
try: return self.bl_info.get('location')
return self.bl_info['location']
except KeyError:
return None
@property @property
def support(self) -> str: def support(self) -> str:
"""Get support from bl_info""" """Get support from bl_info"""
try: return self.bl_info.get('support')
return self.bl_info['support']
except KeyError:
return None
@property @property
def warning(self) -> str: def warning(self) -> str:
"""Get warning from bl_info""" """Get warning from bl_info"""
try: return self.bl_info.get('warning')
return self.bl_info['warning']
except KeyError:
return None
@property @property
def wiki_url(self) -> str: def wiki_url(self) -> str:
"""Get wiki_url from bl_info""" """Get wiki_url from bl_info"""
try: return self.bl_info.get('wiki_url')
return self.bl_info['wiki_url']
except KeyError:
return None
@property @property
def tracker_url(self) -> str: def tracker_url(self) -> str:
"""Get tracker_url from bl_info""" """Get tracker_url from bl_info"""
try: return self.bl_info.get('tracker_url')
return self.bl_info['tracker_url']
except KeyError:
return None
# }}} # }}}
# @classmethod # @classmethod
@@ -169,7 +179,7 @@ class Package:
try: try:
pkg.bl_info = module.bl_info pkg.bl_info = module.bl_info
except AttributeError as err: except AttributeError as err:
raise BadAddon("Module does not appear to be an addon; no bl_info attribute") from err raise exceptions.BadAddon("Module does not appear to be an addon; no bl_info attribute") from err
return pkg return pkg
def download(self, dest: Path, progress_callback=None) -> Path: def download(self, dest: Path, progress_callback=None) -> Path:
@@ -193,10 +203,105 @@ class Package:
utils.install(downloaded, dest_dir) utils.install(downloaded, dest_dir)
def __eq__(self, other):
return self.name == other.name and self.version == other.version
def __lt__(self, other):
return self.version < other.version
def __hash__(self):
return hash((self.name, self.version))
def __repr__(self) -> str: def __repr__(self) -> str:
# return self.name # return self.name
return "Package('name': {}, 'version': {})".format(self.name, self.version) return "Package('name': {}, 'version': {})".format(self.name, self.version)
class ConsolidatedPackage:
"""
Stores a grouping of different versions of the same package
"""
log = logging.getLogger(__name__ + ".ConsolidatedPackage")
def __init__(self, pkg=None):
self.versions = []
self.updateable = False
if pkg is not None:
self.add_version(pkg)
@property
def installed(self) -> bool:
"""Return true if any version of this package is installed"""
for pkg in self.versions:
if pkg.installed:
return True
return False
@property
def name(self) -> str:
"""
Return name of this package. All package versions in a
ConsolidatedPackage should have the same name by definition
Returns None if there are no versions
"""
try:
return self.versions[0].name
except IndexError:
return None
def get_latest_installed_version(self) -> Package:
"""
Return the installed package with the highest version number.
If no packages are installed, return None.
"""
#self.versions is always sorted newer -> older, so we can just grab the first we find
for pkg in self.versions:
if pkg.installed:
return pkg
return None
def get_latest_version(self) -> Package:
"""Return package with highest version number, returns None if there are no versions"""
try:
return self.versions[0] # this is always sorted with the highest on top
except IndexError:
return None
def get_display_version(self) -> Package:
"""
Return installed package with highest version number.
If no version is installed, return highest uninstalled version.
"""
pkg = self.get_latest_installed_version()
if pkg is None:
pkg = self.get_latest_version()
return pkg
def add_version(self, newpkg: Package):
"""Adds a package to the collection of versions"""
if self.name and newpkg.name != self.name:
raise exceptions.PackageException("Name mismatch, refusing to add %s to %s" % (newpkg, self))
for pkg in self:
if pkg == newpkg:
pkg.repositories
if newpkg.installed:
pkg.set_installed_metadata(newpkg)
break
self.versions.append(newpkg)
self.versions.sort(key=lambda v: v.version, reverse=True)
def __iter__(self):
return (pkg for pkg in self.versions)
def __repr__(self):
return ("ConsolidatedPackage<name={}>".format(self.name))
class Repository: class Repository:
""" """
Stores repository metadata (including packages) Stores repository metadata (including packages)
@@ -212,15 +317,22 @@ class Repository:
# def cleanse_packagelist(self): # def cleanse_packagelist(self):
# """Remove empty packages (no bl_info), packages with no name""" # """Remove empty packages (no bl_info), packages with no name"""
def refresh(self): def refresh(self, storage_path: Path, progress_callback=None):
""" """
Requests repo.json from URL and embeds etag/last-modification headers Requests repo.json from URL and embeds etag/last-modification headers
""" """
import requests import requests
if progress_callback is None:
progress_callback = lambda x: None
progress_callback(0.0)
if self.url is None: if self.url is None:
raise ValueError("Cannot refresh repository without a URL") raise ValueError("Cannot refresh repository without a URL")
url = utils.add_repojson_to_url(self.url)
self.log.debug("Refreshing repository from %s", self.url) self.log.debug("Refreshing repository from %s", self.url)
req_headers = {} req_headers = {}
@@ -235,18 +347,18 @@ class Repository:
pass pass
try: try:
resp = requests.get(self.url, headers=req_headers, timeout=60) resp = requests.get(url, headers=req_headers, timeout=60)
except requests.exceptions.InvalidSchema as err: except requests.exceptions.InvalidSchema as err:
raise exceptions.DownloadException("Invalid schema. Did you mean to use http://?") from err raise exceptions.DownloadException("Invalid schema. Did you mean to use http://?") from err
except requests.exceptions.ConnectionError as err: except requests.exceptions.ConnectionError as err:
raise exceptions.DownloadException("Failed to connect. Are you sure '%s' is the correct URL?" % self.url) from err raise exceptions.DownloadException("Failed to connect. Are you sure '%s' is the correct URL?" % url) from err
except requests.exceptions.RequestException as err: except requests.exceptions.RequestException as err:
raise exceptions.DownloadException(err) from err raise exceptions.DownloadException(err) from err
try: try:
resp.raise_for_status() resp.raise_for_status()
except requests.HTTPError as err: except requests.HTTPError as err:
self.log.error('Error downloading %s: %s', self.url, err) self.log.error('Error downloading %s: %s', url, err)
raise exceptions.DownloadException(resp.status_code, resp.reason) from err raise exceptions.DownloadException(resp.status_code, resp.reason) from err
if resp.status_code == requests.codes.not_modified: if resp.status_code == requests.codes.not_modified:
@@ -265,17 +377,22 @@ class Repository:
self.log.debug("Found headers: %s", resp_headers) self.log.debug("Found headers: %s", resp_headers)
progress_callback(0.7)
try: try:
repodict = resp.json() repodict = resp.json()
except json.decoder.JSONDecodeError: except json.decoder.JSONDecodeError:
self.log.exception("Failed to parse downloaded repository") self.log.exception("Failed to parse downloaded repository")
raise exceptions.BadRepositoryException( raise exceptions.BadRepositoryException(
"Could not parse repository downloaded from '%s'. Are you sure this is the correct URL?" % self.url "Could not parse repository downloaded from '%s'. Are you sure this is the correct URL?" % url
) )
repodict['_headers'] = resp_headers repodict['_headers'] = resp_headers
repodict['url'] = self.url
self.set_from_dict(repodict) self.set_from_dict(repodict)
self.to_file(storage_path / utils.format_filename(self.name, ".json"))
progress_callback(1.0)
def to_dict(self, sort=False, ids=False) -> dict: def to_dict(self, sort=False, ids=False) -> dict:
@@ -304,17 +421,20 @@ class Repository:
Get repository attributes from a dict such as produced by `to_dict` Get repository attributes from a dict such as produced by `to_dict`
""" """
def initialize(item, value): # def initialize(item, value):
if item is None: # if item is None:
return value # return value
else: # else:
return item # return item
#Be certain to initialize everything; downloaded packagelist might contain null values #Be certain to initialize everything; downloaded packagelist might contain null values
name = initialize(repodict.get('name'), "") # url = initialize(repodict.get('url'), "")
url = initialize(repodict.get('url'), "") # packages = initialize(repodict.get('packages'), [])
packages = initialize(repodict.get('packages'), []) # headers = initialize(repodict.get('_headers'), {})
headers = initialize(repodict.get('_headers'), {}) name = repodict.get('name', "")
url = repodict.get('url', "")
packages = repodict.get('packages', [])
headers = repodict.get('_headers', {})
self.name = name self.name = name
self.url = url self.url = url
@@ -337,10 +457,26 @@ class Repository:
if self.packages is None: if self.packages is None:
self.log.warning("Writing an empty repository") self.log.warning("Writing an empty repository")
self.log.debug("URL is %s", self.url)
with path.open('w', encoding='utf-8') as repo_file: with path.open('w', encoding='utf-8') as repo_file:
json.dump(self.to_dict(), repo_file, indent=4, sort_keys=True) json.dump(self.to_dict(), repo_file, indent=4, sort_keys=True)
self.log.debug("Repository written to %s" % path) self.log.debug("Repository written to %s" % path)
# def set_from_file(self, path: Path):
# """
# Set the current instance's attributes from a json file
# """
# repo_file = path.open('r', encoding='utf-8')
#
# with repo_file:
# try:
# self.set_from_dict(json.load(repo_file))
# except Exception as err:
# raise BadRepository from err
#
# self.log.debug("Repository read from %s", path)
@classmethod @classmethod
def from_file(cls, path: Path): def from_file(cls, path: Path):
""" """
@@ -351,8 +487,13 @@ class Repository:
with repo_file: with repo_file:
try: try:
repo = cls.from_dict(json.load(repo_file)) repo = cls.from_dict(json.load(repo_file))
except Exception as err: except json.JSONDecodeError as err:
raise BadRepository from err raise exceptions.BadRepositoryException(err) from err
if repo.url is None or len(repo.url) == 0:
raise exceptions.BadRepositoryException("Repository missing URL")
cls.log.debug("Repository read from %s", path) cls.log.debug("Repository read from %s", path)
return repo return repo
def __repr__(self):
return "Repository({}, {})".format(self.name, self.url)

39
package_manager/bpkg/utils.py
View File

@@ -4,6 +4,18 @@ import shutil
import logging import logging
def format_filename(s: str, ext=None) -> str:
"""Take a string and turn it into a reasonable filename"""
import string
if ext is None:
ext = ""
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
filename = ''.join(char for char in s if char in valid_chars)
filename = filename.replace(' ','_')
filename.lower()
filename += ext
return filename
def download(url: str, destination: Path, progress_callback=None) -> Path: def download(url: str, destination: Path, progress_callback=None) -> Path:
""" """
Downloads file at the given url, and if progress_callback is specified, Downloads file at the given url, and if progress_callback is specified,
@@ -17,12 +29,11 @@ def download(url: str, destination: Path, progress_callback=None) -> Path:
log = logging.getLogger('%s.download' % __name__) log = logging.getLogger('%s.download' % __name__)
if progress_callback is None: if progress_callback is None:
# assing to do nothing function # assign to do-nothing function
progress_callback = lambda x: None progress_callback = lambda x: None
progress_callback(0) progress_callback(0)
# derive filename from url if `destination` is an existing directory, otherwise use `destination` directly # derive filename from url if `destination` is an existing directory, otherwise use `destination` directly
if destination.is_dir(): if destination.is_dir():
# TODO: get filename from Content-Disposition header, if available. # TODO: get filename from Content-Disposition header, if available.
@@ -35,10 +46,10 @@ def download(url: str, destination: Path, progress_callback=None) -> Path:
log.info('Downloading %s -> %s', url, local_fpath) log.info('Downloading %s -> %s', url, local_fpath)
try: # try:
resp = requests.get(url, stream=True, verify=True) resp = requests.get(url, stream=True, verify=True)
except requests.exceptions.RequestException as err: # except requests.exceptions.RequestException as err:
raise exceptions.DownloadException(err) from err # raise exceptions.DownloadException(err) from err
try: try:
resp.raise_for_status() resp.raise_for_status()
@@ -187,3 +198,21 @@ def install(src_file: Path, dest_dir: Path):
raise exceptions.InstallException("Failed to copy file to '%s': %s" % (dest_dir, err)) from err raise exceptions.InstallException("Failed to copy file to '%s': %s" % (dest_dir, err)) from err
log.debug("Installation succeeded") log.debug("Installation succeeded")
# def load_repository(repo_storage_path: Path, repo_name: str) -> Repository:
# """Loads <repo_name>.json from <repo_storage_path>"""
# pass
#
# def download_repository(repo_storage_path: Path, repo_name: str):
# """Loads <repo_name>.json from <repo_storage_path>"""
# pass
# this is done in Repository
def add_repojson_to_url(url: str) -> str:
"""Add `repo.json` to the path component of a url"""
from urllib.parse import urlsplit, urlunsplit
parsed_url = urlsplit(url)
new_path = parsed_url.path + "/repo.json"
return urlunsplit((parsed_url.scheme, parsed_url.netloc, new_path, parsed_url.query, parsed_url.fragment))

4
package_manager/messages.py
View File

@@ -1,4 +1,4 @@
from .bpkg import Repository from .bpkg.types import Repository
class Message: class Message:
"""Superclass for all message sent over pipes.""" """Superclass for all message sent over pipes."""
@@ -33,7 +33,7 @@ class Success(SubprocMessage):
class RepositoryResult(SubprocMessage): class RepositoryResult(SubprocMessage):
"""Sent when an operation returns a repository to be used on the parent process.""" """Sent when an operation returns a repository to be used on the parent process."""
def __init__(self, repository: Repository): def __init__(self, repository_name: str):
self.repository = repository self.repository = repository
class Aborted(SubprocMessage): class Aborted(SubprocMessage):

45
package_manager/subproc.py
View File

@@ -7,9 +7,10 @@ from . import utils
from . import bpkg from . import bpkg
from . import messages from . import messages
from .bpkg import exceptions as bpkg_exs from .bpkg import exceptions as bpkg_exs
from .bpkg.types import (Package, Repository)
import logging import logging
def download_and_install_package(pipe_to_blender, package: bpkg.Package, install_path: Path): def download_and_install_package(pipe_to_blender, package: Package, install_path: Path):
"""Downloads and installs the given package.""" """Downloads and installs the given package."""
log = logging.getLogger(__name__ + '.download_and_install') log = logging.getLogger(__name__ + '.download_and_install')
@@ -29,7 +30,7 @@ def download_and_install_package(pipe_to_blender, package: bpkg.Package, install
pipe_to_blender.send(messages.Success()) pipe_to_blender.send(messages.Success())
def uninstall_package(pipe_to_blender, package: bpkg.Package, install_path: Path): def uninstall_package(pipe_to_blender, package: Package, install_path: Path):
"""Deletes the given package's files from the install directory""" """Deletes the given package's files from the install directory"""
#TODO: move package to cache and present an "undo" button to user, to give nicer UX on misclicks #TODO: move package to cache and present an "undo" button to user, to give nicer UX on misclicks
@@ -44,31 +45,37 @@ def uninstall_package(pipe_to_blender, package: bpkg.Package, install_path: Path
pipe_to_blender.send(messages.Success()) pipe_to_blender.send(messages.Success())
def refresh_repository(pipe_to_blender, repo_storage_path: Path, repository_url: str): def refresh_repositories(pipe_to_blender, repo_storage_path: Path, repository_urls: str, progress_callback=None):
"""Retrieves and stores the given repository""" """Downloads and stores the given repository"""
log = logging.getLogger(__name__ + '.refresh') log = logging.getLogger(__name__ + '.refresh_repository')
repository_url = utils.add_repojson_to_url(repository_url)
repo_path = repo_storage_path / 'repo.json' if progress_callback is None:
if repo_path.exists(): progress_callback = lambda x: None
repo = bpkg.Repository.from_file(repo_path) progress_callback(0.0)
if repo.url != repository_url:
# We're getting a new repository
repo = bpkg.Repository(repository_url)
else:
repo = bpkg.Repository(repository_url)
repos = bpkg.load_repositories(repo_storage_path)
def prog(progress: float):
progress_callback(progress/len(repos))
known_repo_urls = [repo.url for repo in repos]
for repo_url in repository_urls:
if repo_url not in known_repo_urls:
repos.append(Repository(repo_url))
for repo in repos:
log.debug("repo name: %s, url: %s", repo.name, repo.url)
for repo in repos:
try: try:
repo.refresh() repo.refresh(repo_storage_path, progress_callback=prog)
except bpkg_exs.DownloadException as err: except bpkg_exs.DownloadException as err:
pipe_to_blender.send(messages.DownloadError(err)) pipe_to_blender.send(messages.DownloadError(err))
raise log.exception("Download error")
except bpkg_exs.BadRepositoryException as err: except bpkg_exs.BadRepositoryException as err:
pipe_to_blender.send(messages.BadRepositoryError(err)) pipe_to_blender.send(messages.BadRepositoryError(err))
raise log.exception("Bad repository")
repo.to_file(repo_path) # TODO: this always writes even if repo wasn't changed progress_callback(1.0)
pipe_to_blender.send(messages.RepositoryResult(repo))
pipe_to_blender.send(messages.Success()) pipe_to_blender.send(messages.Success())

13
package_manager/utils.py
View File

@@ -1,4 +1,9 @@
import bpy
from . import bpkg
from pathlib import Path from pathlib import Path
import logging
from collections import OrderedDict
def fmt_version(version_number: tuple) -> str: def fmt_version(version_number: tuple) -> str:
"""Take version number as a tuple and format it as a string""" """Take version number as a tuple and format it as a string"""
@@ -7,16 +12,18 @@ def fmt_version(version_number: tuple) -> str:
vstr += "." + str(component) vstr += "." + str(component)
return vstr return vstr
def parse_repository_url(url: str) -> str: def sanitize_repository_url(url: str) -> str:
"""Sanitize repository url""" """Sanitize repository url"""
from urllib.parse import urlsplit, urlunsplit from urllib.parse import urlsplit, urlunsplit
parsed_url = urlsplit(url) parsed_url = urlsplit(url)
new_path = parsed_url.path.rstrip("repo.json") # new_path = parsed_url.path.rstrip("repo.json")
new_path = parsed_url.path
return urlunsplit((parsed_url.scheme, parsed_url.netloc, new_path, parsed_url.query, parsed_url.fragment)) return urlunsplit((parsed_url.scheme, parsed_url.netloc, new_path, parsed_url.query, parsed_url.fragment))
def add_repojson_to_url(url: str) -> str: def add_repojson_to_url(url: str) -> str:
"""Add repo.json to a url""" """Add `repo.json` to the path component of a url"""
from urllib.parse import urlsplit, urlunsplit from urllib.parse import urlsplit, urlunsplit
parsed_url = urlsplit(url) parsed_url = urlsplit(url)
new_path = str(Path(parsed_url.path) / "repo.json") new_path = str(Path(parsed_url.path) / "repo.json")
return urlunsplit((parsed_url.scheme, parsed_url.netloc, new_path, parsed_url.query, parsed_url.fragment)) return urlunsplit((parsed_url.scheme, parsed_url.netloc, new_path, parsed_url.query, parsed_url.fragment))