nrp_cmd.async_client.invenio

 22class AsyncInvenioRepositoryClient(AsyncRepositoryClient):
 23    """An abstract client for NRP repositories.
 24
 25    Usually, subclasses of this class are not instantiated directly in your code.
 26    To get an instance of a repository, use the high-level method `get_async_client`:
 27
 28    ```
 29    my_client = await async_client(config?, url=url, refresh=False/True)
 30    my_client = await async_client(config?, alias=alias, refresh=False/True)
 31    ```
 32
 33    and then use the instance.
 34    """
 35
 36    @classmethod
 37    async def can_handle_repository(
 38        cls, url: URL | str, verify_tls: bool = True
 39    ) -> URL | None:
 40        """Return if this client can handle a repository that contains the passed URL.
 41
 42        This method can make an http request or use any means to check if the repository
 43        at the URL can be handled by this client.
 44
 45        :param url: any url within the repository (root, record api, html, documentation
 46        if running on the same host, ...)
 47        :param verify_tls: whether to verify tls (should be switched on for production)
 48        :return: API url of the server or None if this client can not handle the repository
 49        """
 50        connection = AsyncConnection(verify_tls=verify_tls)
 51        if isinstance(url, str):
 52            url = URL(url)
 53
 54        # check if the repository is an NRP invenio repository
 55        well_known_url = url.with_path("/.well-known/repository")
 56        try:
 57            data = await connection.get(url=well_known_url, result_class=dict[str, Any])
 58            if "invenio_version" in data:
 59                if "api" in data.get("links", {}):
 60                    return URL(data["links"]["api"])
 61                # fallback for older oarepo-runtime versions
 62                return url.with_path("/api")
 63        except Exception:
 64            pass
 65
 66        # check if the repository is a plain Invenio RDM repository, such as zenodo
 67        try:
 68            root_page_data = await connection.get(url=url.with_path("/"), result_class=str)
 69            if '<meta name="generator" content="InvenioRDM' in root_page_data:
 70                return url.with_path("/api")
 71        except Exception:
 72            pass
 73        return None
 74
 75    @classmethod
 76    async def from_configuration(
 77        cls,
 78        config: RepositoryConfig,
 79        refresh: bool = False,
 80        limiter: Limiter | None = None,
 81        extra_tokens: dict[URL, str] | None = None,
 82    ) -> Self:
 83        """Create a client from the given configuration.
 84
 85        :param config: the configuration for the repository
 86        :param refresh: refresh the configuration by calling get_repository_info
 87        """
 88        ret = cls(config=config, limiter=limiter, extra_tokens=extra_tokens)
 89        if refresh or not config.info:
 90            config.info = await ret.get_repository_info(refresh=True)
 91        return ret
 92
 93    def __init__(
 94        self,
 95        config: RepositoryConfig,
 96        limiter: Limiter | None = None,
 97        extra_tokens: dict[URL, str] | None = None,
 98    ):
 99        """Create a new client for the given repository configuration."""
100        self._config = config
101        tokens: dict[URL, str] = {}
102        if extra_tokens:
103            tokens.update(extra_tokens)
104        if config.token:
105            tokens[config.url] = config.token
106        self._connection = AsyncConnection(
107            limiter=limiter,
108            tokens=tokens,
109            verify_tls=config.verify_tls,
110            retry_count=config.retry_count,
111            retry_after_seconds=config.retry_after_seconds,
112        )
113
114    async def get_repository_info(
115        self, refresh: bool = True
116    ) -> RepositoryInfo:
117        """Get information about the repository.
118
119        This call is cached inside the RepositoryConfig instance.
120
121        :param include_models: fetch information about the supported models as well
122        :param refresh: refresh the info from the server
123        """
124        
125        if self._config.info and not refresh:
126            return self._config.info
127
128        try:
129            info = await self._connection.get(
130                url=self._config.well_known_repository_url,
131                result_class=RepositoryInfo,
132            )
133            self._config.info = info
134
135            if not self._config.info.links.models:
136                raise ValueError("The repository does not provide the models link.")
137
138            models = await self._connection.get(
139                url=self._config.info.links.models,
140                result_class=list[ModelInfo],
141            )
142
143            info.models = {
144                model.type: model for model in models
145            }
146
147        except* (RepositoryClientError, RepositoryCommunicationError) as exc:
148            if is_instance_of_exceptions(exc, StructureError):
149                raise exc
150            # not a NRP based repository, suppose that it is plain invenio rdm
151            info = self._config.info = make_rdm_info(self._config.url)
152
153        return info
154
155
156    @property
157    def records(self) -> AsyncInvenioRecordsClient:
158        """Return client for accessing records."""
159        assert self._config.info, "Repository info is not available, can not create records client."
160        return AsyncInvenioRecordsClient(self._connection, self._config.info)