olivia_finder.data_source.librariesio_ds

  1from typing import List, Union
  2from ..utilities.exception import OliviaFinderException
  3from ..utilities.config import Configuration
  4from .data_source import DataSource
  5import sys
  6import time
  7
  8
  9# Load the Libraries.io API key before importing the librariesio package
 10# -----------------------------------------------------------------------
 11
 12# try to load the API key from the configuration file
 13API_KEY = Configuration().get_key('librariesio', 'api_key')
 14if API_KEY is None:
 15    raise OliviaFinderException("API key for Libraries.io not found")
 16
 17# Set as environment variable
 18import os
 19os.environ["LIBRARIES_API_KEY"] = API_KEY
 20from pybraries.search import Search
 21
 22class LibrariesioDataSource(DataSource):
 23    """
 24    Data source for the Libraries.io API
 25
 26    .. warning::
 27
 28        - The API key must be set as an environment variable with the name `LIBRARIES_API_KEY`
 29        - The class will raise an exception if the API key is not found
 30        - The API key can be set in the configuration file
 31        - The API key can be obtained from the user's profile in Libraries.io
 32
 33    """
 34
 35    DEFAULT_DESCRIPTION = "Libraries.io data source"
 36    
 37    def __init__(self, platform: str):
 38        """
 39        Constructor of the class
 40
 41        Parameters
 42        ----------
 43        platform : str
 44            Platform of the data source to search for packages
 45
 46        Raises
 47        ------
 48        LibrariesIoException
 49            If the search object cannot be created
 50        """
 51        super().__init__()
 52        self.platform = platform
 53
 54        # Create the search object
 55        try:
 56            self.search = Search()
 57        except Exception as e:
 58            pass
 59
 60    def obtain_package_names(self) -> List[str]:
 61        """
 62        Obtains the list of packages from the data source.
 63        """
 64        raise NotImplementedError
 65    
 66    def obtain_package_data(self, package_name:str) -> Union[dict, None]:
 67        """
 68        Obtains the data of a package from the data source.
 69
 70        Parameters
 71        ----------
 72        package_name : str
 73            Name of the package
 74
 75        Returns
 76        -------
 77        dict
 78            Dictionary with the package data
 79        None
 80            If the package is not found
 81            
 82        Example
 83        -------
 84        >>> data_source = LibrariesioDataSource('name', 'description', 'platform')
 85        >>> data_source.obtain_package_data('package_name')
 86        {'name': 'package_name', 'version': '1.0.0', 'dependencies': ['package1', 'package2'], 'url': 'www.example.com'}
 87        """
 88
 89        # Get the package version and url
 90        self.logger.debug(f"Obtaining data of {package_name}")
 91
 92        # Redirect the output to /dev/null to avoid printing the output of the search (pybraries bug)
 93        stdout = sys.stdout     # Keep the standard output backed up
 94        sys.stdout = open('/dev/null', 'w')
 95
 96        try:
 97            dependencies_data = self.search.project_dependencies(
 98                platforms=self.platform, project=package_name
 99            )
100
101            # Check if data is empty
102            if dependencies_data is None:
103
104                # Sleep for 3 second to avoid the API rate limit
105                self.logger.debug(f"Package {package_name} not found. Sleeping for 4 seconds for the API rate limit")
106                time.sleep(4)
107
108            # Try again
109            dependencies_data = self.search.project_dependencies(
110                platforms=self.platform, project=package_name
111            )
112
113            # Check if data is empty
114            if dependencies_data is None:
115
116                self.logger.debug(f"Package {package_name} not found")
117                return None
118            else:
119                self.logger.debug(f"Package {package_name} found")
120
121        except Exception as e:
122            self.logger.error(f"Exception while obtaining {package_name} dependencies")
123            return None
124
125        finally:
126            # Restore the original standard output
127            sys.stdout.close()
128            sys.stdout = stdout 
129
130        version = dependencies_data["dependencies_for_version"]
131        url = dependencies_data["package_manager_url"]
132
133
134        # Obtain the dependencies (dependencies field, project_name)
135        # Obtain the dependency version (atest_stable field)
136        dependencies = []
137
138        for dependency in dependencies_data["dependencies"]:
139            dependency_name = dependency["project_name"]
140            dependency_version = dependency["latest_stable"]
141
142            dependency_data = {
143                "name": dependency_name,
144                "version": dependency_version,
145            }
146
147            dependencies.append(dependency_data)
148
149        return {
150            "name": package_name,
151            "version": version,
152            "dependencies": dependencies,
153            "url": url
154        }
155        
156    def obtain_packages_data(self, package_names: List[str]) -> List[dict]:
157        '''
158        Obtains the data of a list of package names from the data source.
159
160        Returns
161        -------
162        List[dict]
163            The data of the packages as a list of dictionaries
164
165        Example
166        -------
167        >>> data_source = LibrariesioDataSource('name', 'description', 'platform')
168        >>> data_source.obtain_packages_data(['package1', 'package2'])
169        [{'name': 'package1', 'version': '1.0.0', 'dependencies': ['package1', 'package2'], 'url': 'www.example.com'}, 
170        {'name': 'package2', 'version': '1.0.0', 'dependencies': ['package1', 'package2'], 'url': 'www.example.com'}]
171        '''
172
173        packages_data = []
174
175        for package_name in package_names:
176            if package_data := self.obtain_package_data(package_name):
177                packages_data.append(package_data)
178
179        return packages_data