Source code for ocdsextensionregistry.extension_registry

r"""
Filter the versions of extensions in the registry, and access information about matching versions.

.. code:: python

    from ocdsextensionregistry import ExtensionRegistry

    extensions_url = 'https://raw.githubusercontent.com/open-contracting/extension_registry/main/extensions.csv'
    extension_versions_url = 'https://raw.githubusercontent.com/open-contracting/extension_registry/main/extension_versions.csv'

    registry = ExtensionRegistry(extension_versions_url, extensions_url)
    for version in registry.filter(core=True, version='v1.1.4', category='tender'):
        print(f'The {version.metadata[name][en]} extension ("{version.id}") is maintained at {version.repository_html_page}')
        print(f'Run `git clone {version.repository_url}` to make a local copy in a {version.repository_name} directory')
        print(f'Get its patch at {version.base_url}release-schema.json\n')

Output::

    The Enquiries extension ("enquiries") is maintained at https://github.com/open-contracting-extensions/ocds_enquiry_extension
    Run `git clone git@github.com:open-contracting-extensions/ocds_enquiry_extension.git` to make a local copy in a ocds_enquiry_extension directory
    Get its patch at https://raw.githubusercontent.com/open-contracting-extensions/ocds_enquiry_extension/v1.1.4/release-schema.json

To work with the files within a version of an extension:

-  :meth:`~ocdsextensionregistry.extension_version.ExtensionVersion.metadata` parses and provides consistent access to the information in ``extension.json``
-  :meth:`~ocdsextensionregistry.extension_version.ExtensionVersion.schemas` returns the parsed contents of schema files
-  :meth:`~ocdsextensionregistry.extension_version.ExtensionVersion.codelists` returns the parsed contents of codelist files (see more below)
-  :meth:`~ocdsextensionregistry.extension_version.ExtensionVersion.files` returns the unparsed contents of all files

See additional details in :doc:`extension_version`.
"""  # noqa: E501

import csv
from io import StringIO
from urllib.parse import urlsplit

from ocdsextensionregistry.exceptions import DoesNotExist, MissingExtensionMetadata
from ocdsextensionregistry.extension import Extension
from ocdsextensionregistry.extension_version import ExtensionVersion
from ocdsextensionregistry.util import _resolve




class ExtensionRegistry:


    def __init__(self, extension_versions_data, extensions_data=None):
        """
        Accept ``extension_versions.csv`` and, optionally, ``extensions.csv`` as either URLs or data (as string) and
        read them into :class:`~ocdsextensionregistry.extension_version.ExtensionVersion` objects.

        If extensions_data is not provided, the extension versions will not have category or core properties.
        URLs starting with ``file://`` will be read from the filesystem.
        """
        self.versions = []

        # If extensions data is provided, prepare to merge it with extension versions data.
        extensions = {}
        if extensions_data:
            extensions_data = _resolve(extensions_data)
            for row in csv.DictReader(StringIO(extensions_data)):
                extension = Extension(row)
                extensions[extension.id] = extension

        extension_versions_data = _resolve(extension_versions_data)
        for row in csv.DictReader(StringIO(extension_versions_data)):
            version = ExtensionVersion(row)
            if version.id in extensions:
                version.update(extensions[version.id])
            self.versions.append(version)




    def filter(self, **kwargs):
        """
        Return the extension versions in the registry that match the keyword arguments.

        :raises MissingExtensionMetadata: if the keyword arguments refer to extensions data, but the extension registry
                                          was not initialized with extensions data
        """
        try:
            return [ver for ver in self.versions if all(getattr(ver, k) == v for k, v in kwargs.items())]
        except AttributeError as e:
            self._handle_attribute_error(e)




    def get(self, **kwargs):
        """
        Return the first extension version in the registry that matches the keyword arguments.

        :raises DoesNotExist: if no extension version matches
        :raises MissingExtensionMetadata: if the keyword arguments refer to extensions data, but the extension registry
                                          was not initialized with extensions data
        """
        try:
            return next(ver for ver in self.versions if all(getattr(ver, k) == v for k, v in kwargs.items()))
        except StopIteration as e:
            raise DoesNotExist(f"Extension version matching {kwargs!r} does not exist.") from e
        except AttributeError as e:
            self._handle_attribute_error(e)




    def get_from_url(self, url):
        """
        Return the first extension version in the registry whose base URL matches the given URL.

        :raises DoesNotExist: if no extension version matches
        """
        parsed = urlsplit(url)
        path = f"{parsed.path.rsplit('/', 1)[0]}/"
        return self.get(base_url=parsed._replace(path=path).geturl())




    def __iter__(self):
        """Iterate over the extension versions in the registry."""
        yield from self.versions


    def _handle_attribute_error(self, e):
        if "'category'" in str(e.args) or "'core'" in str(e.args):
            raise MissingExtensionMetadata("ExtensionRegistry must be initialized with extensions data.") from e
        raise  # noqa: PLE0704 # false positive