# -*- coding: utf-8 -*-
"""Main module."""
import copy
import logging
import os
import sys
from cerberus import Validator
from jinja2 import Environment
# import yaml
from stevedore import driver
from frkl import download_cached_file
from frutils import is_url_or_abbrev
from .lupkg_index import LupkgMultiIndex, LupkgFileIndex
log = logging.getLogger("lucify")
DEFAULT_PKG_TYPE = "default"
ENV = Environment()
INDEX_DESC_SCHEMA = {
"alias": {"type": "string"},
"url": {"type": "string"},
"base_url": {"type": "string"},
"type": {"type": "string"},
"init_params": {"type": "dict", "allow_unknown": True},
}
INDEX_DESC_VALIDATOR = Validator(INDEX_DESC_SCHEMA)
# =============================================================
# global index management
[docs]def create_index_desc(url_orig):
remote = is_url_or_abbrev(url_orig)
if remote:
if url_orig.endswith(".git"):
raise Exception("Remote index folders not supported yet")
is_file = True
url = download_cached_file(url_orig)
else:
real_path = os.path.realpath(os.path.expanduser(url_orig))
if not os.path.exists(real_path):
raise Exception("Index url does not exist: {}".format(url_orig))
if os.path.isfile(real_path):
is_file = True
else:
is_file = False
url = url_orig
if remote:
index_type = "file"
base_url = os.path.dirname(url)
else:
if is_file:
index_type = "file"
base_url = os.path.dirname(url)
else:
index_type = "folder"
base_url = url
desc = {}
desc["alias"] = url_orig
desc["url"] = url
desc["base_url"] = base_url
desc["type"] = index_type
desc["init_params"] = {}
return desc
[docs]def create_index(index_desc_dict):
log.debug("Creating index for description: {}".format(index_desc_dict))
# print("Creating index for description: {}".format(index_desc_dict))
valid = INDEX_DESC_VALIDATOR(index_desc_dict)
if not valid:
errors = copy.deepcopy(INDEX_DESC_VALIDATOR.errors)
raise Exception(
"Can't parse index description '{}': {}".format(index_desc_dict, errors)
)
index_type = index_desc_dict["type"]
url = index_desc_dict["url"]
base_url = index_desc_dict["base_url"]
init_params = copy.deepcopy(index_desc_dict["init_params"])
init_params["url"] = url
init_params["pkg_base_url"] = base_url
init_params["alias"] = index_desc_dict.get("alias", None)
if index_type == "multi":
raise Exception("Dynamically creating multi-indexes not supported yet")
log2 = logging.getLogger("stevedore")
out_hdlr = logging.StreamHandler(sys.stdout)
out_hdlr.setFormatter(
logging.Formatter("dictlet_finder plugin error -> %(message)s")
)
out_hdlr.setLevel(logging.DEBUG)
log2.addHandler(out_hdlr)
log2.setLevel(logging.INFO)
log.debug("Loading dictlet_finder...")
mgr = driver.DriverManager(
namespace="lupkg.indexes",
name=index_type,
invoke_on_load=True,
invoke_kwds=init_params,
)
log.debug(
"Registered lupkg index: {}".format(
", ".join(ext.name for ext in mgr.extensions)
)
)
return mgr.driver
DEFAULT_INDEX_NAME = "default"
[docs]class LupkgIndexes(object):
DEFAULT_INDEX_CONFIG_PATH = os.path.expanduser("~/.local/share/lupkg/indexes")
DEFAULT_INDEX_READER_PARAMS = {"use_files": False, "use_metadata_files": True}
def __init__(self):
self.lupkg_path = None
self.index_cache = {}
self.index = None
[docs] def list_available_index_names(self):
return self.get_indexes_index().get_pkg_names()
[docs] def download_index(self, index_name, ignore_if_already_installed=True):
log.debug("Downloading index: {}".format(index_name))
result = self.get_local().install_pkg(
index_name,
index="indexes",
ignore_if_already_installed=ignore_if_already_installed,
install_properties={"force": True},
)
self.get_local(update=True)
return result
[docs] def get_indexes_index(self, update=False):
if self.index is None:
self.index = LupkgFileIndex(
url="gl:lupkg/index/indexes.index", alias="indexes"
)
if update:
self.index.update()
self.index_cache = {}
return self.index
[docs] def get_local(self, update=False):
if self.lupkg_path is None or update:
from .lupkg_list import LupkgPath
log.debug(
"(Re-)Loading path: {}".format(LupkgIndexes.DEFAULT_INDEX_CONFIG_PATH)
)
self.lupkg_path = LupkgPath(LupkgIndexes.DEFAULT_INDEX_CONFIG_PATH)
return self.lupkg_path
[docs] def update_index(self, index_alias_or_url):
self.get_indexes_index(update=True)
self.get_index(index_alias_or_url, update=True)
[docs] def get_index(self, index_alias_or_url, update=False):
if index_alias_or_url == "indexes":
return self.get_indexes_index()
if index_alias_or_url in self.index_cache.keys() and not update:
return self.index_cache[index_alias_or_url]
if index_alias_or_url in self.get_indexes_index().get_pkg_names():
if update:
self.download_index(
index_alias_or_url, ignore_if_already_installed=False
)
else:
self.download_index(index_alias_or_url)
pkg = self.get_local().get_pkg(index_alias_or_url)
files = pkg.get_file_names()
indexes = []
for f in files:
path = pkg.get_path_for_file(f)
index_desc = create_index_desc(path)
index = create_index(index_desc)
indexes.append(index)
index = LupkgMultiIndex(
url=None, pkg_base_url=None, alias=index_alias_or_url, indexes=indexes
)
else:
index_desc = create_index_desc(index_alias_or_url)
log.debug("Created index description: {}".format(index_desc))
index = create_index(index_desc)
self.index_cache[index_alias_or_url] = index
return index
LUPKG_INDEXES = LupkgIndexes()