Source code for citelang.main.package

__author__ = "Vanessa Sochat"
__copyright__ = "Copyright 2022, Vanessa Sochat"
__license__ = "MPL 2.0"

import citelang.main.cache as cache
import citelang.main.endpoints as endpoints
import citelang.main.packages as packages
from citelang.logger import logger


[docs]def get_package( manager, name, version=None, data=None, use_cache=True, manager_kwargs=None, ): """ Return a package handle for a custom package (defined in citelang) or libraries.io """ # Custom kwargs for the manager manager_kwargs = manager_kwargs or {} # Case 1: a package manager provided by citelang if manager in packages.manager_names: return CustomPackage( name=name, manager=manager, version=version, data=data, use_cache=use_cache, manager_kwargs=manager_kwargs, ) # Case 2: libraries.io return Package( name=name, manager=manager, version=version, data=data, use_cache=use_cache, )
[docs]class PackageBase: def __init__( self, manager, name, version=None, data=None, use_cache=True, manager_kwargs=None, ): """ The version can be set explicitly or come from the name. """ self.manager = manager self.version = version self.parse(name) self.data = data or {} self.latest = None self.use_cache = use_cache self.cache = cache.cache self.manager_kwargs = manager_kwargs or {} self.underlying_manager = None def __repr__(self): return str(self) def __str__(self): return "[citelang-package]" @property def cache_name(self): return f"package/{self.manager}/{self.name}" @property def homepage(self): url = self.data.get("package", {}).get("homepage") if not url and self.manager == "pypi": return f"https://pypi.org/project/{self.name}" if not url and self.manager == "go": return f"https://{self.name}" # Some R urls have TWO and they are wonky if url and " " in url: url = url.strip().split(" ")[0] return url
[docs] def parse(self, name): """ Parse a manager and name into the package """ if "@" in name: name, version = name.split("@", 1) self.version = version if "[" in name and "]" in name: name = name.split("[", 1)[0] # invalid characters found! name = name.replace(";", "") self.name = name
[docs] def dependencies(self, return_data=False): raise NotImplementedError
[docs] def info(self): raise NotImplementedError
[docs]class CustomPackage(PackageBase): """ A wrapper for a custom package (provided by citelang) """ def __init__( self, manager, name, version=None, data=None, use_cache=True, manager_kwargs=None, ): super().__init__(manager, name, version, data, use_cache, manager_kwargs) if not self.underlying_manager: self.underlying_manager = packages.managers[self.manager]( **self.manager_kwargs ) @property def homepage(self): url = self.data.get("homepage") or self.data.get("package", {}).get("homepage") if not url and self.manager == "github": return f"https://github.com/{self.name}" if not url and self.manager in ["go.mod", "go"]: return f"https://{self.name}" if url and " " in url: url = url.strip().split(" ")[0] return url
[docs] def dependencies(self, return_data=False): """ Retrieve custom package dependencies """ manager = self.underlying_manager # Use manager cache first if "dependencies" in manager.data: return manager.data["dependencies"] cache_name = None result = None # If we aren't given a version, look for default branches if not self.version and manager.default_versions and self.use_cache: for version in manager.default_versions: cache_name = f"package/{self.manager}/{self.name}/{version}" result = self.cache.get(cache_name) if result: result = endpoints.get_endpoint("dependencies", data=result) self.version = version break # If we don't have a version, we have to retrieve an update if not result and not self.version: deps = manager.package(self.name) result = endpoints.get_endpoint("dependencies", data=deps) self.version = result.data.get("default_version") cache_name = f"package/{self.manager}/{self.name}/{self.version}" # We have a version, either retrieve from cache or anew elif not result and self.version: cache_name = f"package/{self.manager}/{self.name}/{self.version}" result = self.cache.get(cache_name) if not result or not self.use_cache: version = "@%s" % self.version if self.version else "" logger.info("Retrieving new result for %s%s..." % (self.name, version)) deps = manager.package(self.name) result = endpoints.get_endpoint("dependencies", data=deps) elif result: result = endpoints.get_endpoint("dependencies", data=result) # If cache is enabled, we save the result if result and cache_name: self.cache.set(cache_name, result) # Return the wrapped result or raw data deps = result.data.get("dependencies", []) self.data["package"] = result.data self.data["dependencies"] = deps if return_data: return deps return result
[docs] def info(self): """ Get info for a custom package """ manager = self.underlying_manager # Use manager cache first if "package" in manager.data: return manager.data["package"] # First try retrieving from the cache result = self.cache.get(self.cache_name) if not result or not self.use_cache: version = "@%s" % self.version if self.version else "" logger.info( "Retrieving new result for package %s%s..." % (self.name, version) ) pkg = manager.package(self.name) result = endpoints.get_endpoint( "package", data=pkg, package_name=self.name, manager=self.manager ) else: result = endpoints.get_endpoint( "package", data=result, package_name=self.name, manager=self.manager ) # If cache is enabled, we save the result self.cache.set(self.cache_name, result) self.data["package"] = result.data return result
[docs]class Package(PackageBase): """ A basic wrapper for package parsing functions (libraries.io) """
[docs] def dependencies(self, return_data=False): """ Retrieve libraries.io dependencies """ if not self.version: self.info() self.version = self.latest # Check for result in cache cache_name = f"dependencies/{self.manager}/{self.name}/{self.version}" result = self.cache.get(cache_name) if not result or not self.use_cache: logger.info( "Retrieving new result for %s@%s dependencies..." % (self.name, self.version) ) result = endpoints.get_endpoint( "dependencies", manager=self.manager, package_name=self.name, version=self.version, ) else: result = endpoints.get_endpoint("dependencies", data=result) # If cache is enabled, we save the result self.cache.set(cache_name, result) deps = result.data.get("dependencies", []) self.data["dependencies"] = deps if return_data: return deps return result
[docs] def info(self): """ Get info for a libraries.io package """ result = self.cache.get(self.cache_name) if not result and not self.manager: return result if not result or not self.use_cache: version = "@%s" % self.version if self.version else "" logger.info( "Retrieving new result for package %s%s..." % (self.name, version) ) result = endpoints.get_endpoint( "package", manager=self.manager, package_name=self.name, ) else: result = endpoints.get_endpoint( "package", data=result, manager=self.manager, package_name=self.name ) # Set a latest version if we find one if "versions" in result.data and result.data["versions"]: self.latest = result.data["versions"][-1]["number"] self.cache.set(self.cache_name, result) self.data["package"] = result.data return result