# -*- coding: utf-8 -*-
"""Main module."""
import abc
import logging
import os
from collections import OrderedDict
from cerberus import Validator
from jinja2 import Environment
# import yaml
from ruamel.yaml import YAML
from ruamel.yaml.comments import CommentedMap
from six import string_types
from frkl import dict_from_url
from luci.finders import FolderFinder
from luci.lucify import Lucifier
from luci.readers import LupkgFolderReader
from .lupkg import LuPKG, LupkgMetadataException
log = logging.getLogger("lucify")
DEFAULT_PKG_TYPE = "default"
ENV = Environment()
INDEX_DESC_SCHEMA = {
"url": {"type": "string"},
"base_url": {"type": "string"},
"type": {"type": "string"},
"init_params": {"type": "dict", "allow_unknown": True},
}
INDEX_DESC_VALIDATOR = Validator(INDEX_DESC_SCHEMA)
LUPKG_CACHE_BASE = os.path.expanduser("~/.local/share/lupkg/cache")
[docs]class LupkgIndex(object):
"""Base class to hold information about a package repository.
This can be extended by classes that wish to implement a specific
way of storing and querying that data (e.g. using a database etc.).
"""
# @classmethod
# def create(cls, *urls):
#
# descs = expand_index_url(urls)
# if len(descs) == 0:
# raise Exception("Could not find any indexes for url(s): {}".format(urls))
# elif len(descs) == 1:
# index = create_index(descs[0])
# else:
# indexes = []
# for d in descs:
# i = create_index(d)
# indexes.append(i)
#
# index = LupkgMultiIndex(None, None, indexes)
#
# return index
def __init__(self, url=None, alias=None, pkg_base_url=None):
if alias is None:
alias = url
self.name = alias
self.url = url
if pkg_base_url is None:
pkg_base_url = url
self.pkg_base_url = pkg_base_url
self.cached_pkgs = {}
[docs] def get_name(self):
return self.name
[docs] def update(self):
self.cached_pkgs = {}
self.update_index()
[docs] @abc.abstractmethod
def update_index(self):
"""Updates the index."""
pass
[docs] @abc.abstractmethod
def get_package_types(self):
"""Returns a list of package types in this repository.
Returns:
list: the list of package types
"""
pass
[docs] @abc.abstractmethod
def get_available_packages(self):
"""Returns all available package names in no particular order.
Returns:
list: a list of package names
"""
pass
[docs] def get_pkg_names(self):
"""Returns a sorted list of available package names.
Returns:
list: a list of package names
"""
return sorted(self.get_available_packages())
[docs] def save_index_file(self, path):
yaml = YAML()
yaml.default_flow_style = False
with open(path, "w") as fi:
yaml.dump(self.get_all_metadata(), fi)
[docs] def get_pkgs_with_file(self, file):
"""Returns a list of package names that contain a file with the specified name."""
result = set()
for pn in self.get_pkg_names():
pkg = self.get_pkg(pn)
if pkg.has_file(file):
result.add(pn)
return list(result)
[docs] def get_pkg(self, name):
"""Returns the :class:`~LuPKG` object for the specified package name.
Return:
LuPKG: the package details, or none if package doesn't exist
"""
if name not in self.cached_pkgs.keys():
pkg_details = self.get_pkg_metadata(name)
if not pkg_details:
self.cached_pkgs[name] = None
else:
pkg = LuPKG(pkg_details, base_url=self.pkg_base_url)
self.cached_pkgs[name] = pkg
result = self.cached_pkgs[name]
return result
[docs] def get_all_urls(self, properties=None):
"""Returns a map of all package names with their corresponding url for the latest version.
Args:
properties (dict): optional properties to be used in the package selection
Returns:
OrderedDict: a dict with the package name as key, and the url as value
"""
result = OrderedDict()
for pn in self.get_pkg_names():
pkg = self.get_pkg(pn)
# pkg = self.repos.pkg_descs.get(pn, None)
try:
url = pkg.get_url_details(properties=properties)
result[pn] = url
except (LupkgMetadataException) as e:
log.error(e)
return result
[docs]class LupkgFileIndex(LupkgIndex):
""":class:`LuPKGS` implementation that reads one or multiple yaml files to get package metadata.
"""
def __init__(self, url=None, alias=None, pkg_base_url=None, **kwargs):
if not isinstance(url, string_types):
raise Exception(
"Index file needs to be of type string, instead it is '{}'".format(
type(url)
)
)
if pkg_base_url is None:
pkg_base_url = os.path.dirname(url)
if alias is None:
alias = url
super(LupkgFileIndex, self).__init__(
url=url, alias=alias, pkg_base_url=pkg_base_url, **kwargs
)
self.get_index()
[docs] def update_index(self):
return self.get_index(update=True)
[docs] def get_index(self, update=False):
log.debug("Updating index: {}".format(self.url))
# print("Updating index: {}".format(self.url))
self.metadata = dict_from_url(
self.url, update=update, cache_base=LUPKG_CACHE_BASE
)
[docs] def get_available_packages(self):
return self.metadata.keys()
[docs]class LupkgMultiIndex(LupkgIndex):
""":class:`LuPKGS` implementation to hold multiple, different LuPKGS objects
Args:
name (str): the name of this index
lupkgs (list): a list of LupkgIndex objects
**kwargs (dict): additional init values
"""
def __init__(self, url=None, alias=None, pkg_base_url=None, indexes=None, **kwargs):
# url/base_url is not used in this one
if indexes is None:
raise Exception("No child indexes specified.")
if not isinstance(alias, string_types):
raise Exception("No alias provided.")
super(LupkgMultiIndex, self).__init__(
url=url, alias=alias, pkg_base_url=pkg_base_url, **kwargs
)
self.packages = []
self.indexes = indexes
for index in self.indexes:
self.packages.extend(index.get_pkg_names())
[docs] def update_index(self):
self.packages = []
for i in self.indexes:
i.update_index()
self.packages.extend(i.get_pkg_names())
[docs] def get_available_packages(self):
return self.packages
[docs]class LupkgFolderLucifier(Lucifier):
def __init__(self, reader_params=None, **kwargs):
super(LupkgFolderLucifier, self).__init__("repo", **kwargs)
if reader_params is None:
reader_params = {}
self.reader = LupkgFolderReader(**reader_params)
self.finder = FolderFinder()
self.pkg_descs = CommentedMap()
[docs] def get_default_dictlet_reader(self):
return self.reader
[docs] def get_default_dictlet_finder(self):
return self.finder
[docs] def process_dictlet(self, metadata, dictlet_details=None):
for pkg_name, details in metadata.items():
if pkg_name in self.pkg_descs.keys():
log.warn(
"Duplicate description for package '{}', overwriting existing one...".format(
pkg_name
)
)
self.pkg_descs[pkg_name] = details
[docs]class LupkgFolderIndex(LupkgIndex):
""":class:`~LuPKGS` implementation that stores metadata about packages
in a folder structure.
Internally this uses the `luci <https://github.com/makkus/luci>`_
python library to read the metadata from the folder. This happens
in the :class:`~LupkgMetadataFolderLucifier` class to hold the relevant metadata
derived from the folder.
"""
def __init__(
self, url=None, alias=None, pkg_base_url=None, reader_params=None, **kwargs
):
if pkg_base_url is None:
pkg_base_url = url
if alias is None:
alias = url
if reader_params is None:
reader_params = {}
super(LupkgFolderIndex, self).__init__(
url=url, alias=alias, pkg_base_url=pkg_base_url, **kwargs
)
self.reader_params = reader_params
self.update_index()
[docs] def update_index(self):
self.repos = LupkgFolderLucifier(reader_params=self.reader_params)
self.repos.overlay_dictlet(self.url, add_dictlet=True)
self.repos.process()
[docs] def get_available_packages(self):
return self.repos.pkg_descs.keys()