nrp_cmd.async_client
1import importlib.metadata as importlib_metadata 2from functools import lru_cache 3 4from yarl import URL 5 6from ..config import Config 7from . import base_client, connection, invenio, streams 8from .base_client import AsyncRepositoryClient 9from .connection.limiter import Limiter 10 11 12@lru_cache(maxsize=1) 13def async_client_classes() -> list[type[AsyncRepositoryClient]]: 14 """ 15 Load all available asynchronous client classes. 16 """ 17 async_client_classes: list[type[AsyncRepositoryClient]] = [] 18 for ep in importlib_metadata.entry_points(group="nrp_cmd.async_client"): 19 async_client_classes.append(ep.load()) 20 return async_client_classes 21 22 23async def get_async_client( 24 repository: str | URL, 25 refresh: bool = False, 26 limiter: Limiter | None = None, 27 config: Config | None = None, 28) -> AsyncRepositoryClient: 29 """ 30 Get an asynchronous client for the given repository. 31 32 :param repository: the repository alias or URL 33 :param refresh: whether to refresh the client configuration from the server 34 :param max_connections: the maximum number of parallel connections 35 :param config: the configuration to use. If not given, the configuration is loaded 36 from the configuration file. 37 :return: an asynchronous client for the repository 38 """ 39 if not config: 40 config = Config.from_file() 41 42 repository_config = config.find_repository(repository) 43 for async_client_class in async_client_classes(): 44 if await async_client_class.can_handle_repository(repository_config.url): 45 return await async_client_class.from_configuration( 46 repository_config, refresh=refresh, limiter=limiter 47 ) 48 raise ValueError(f"No async client found for repository {repository_config.url}") 49 50 51__all__ = ( 52 "get_async_client", 53 "AsyncRepositoryClient", 54 "connection", 55 "invenio", 56 "streams", 57 "base_client", 58)
async def
get_async_client( repository: str | yarl.URL, refresh: bool = False, limiter: nrp_cmd.async_client.connection.Limiter | None = None, config: nrp_cmd.config.Config | None = None) -> AsyncRepositoryClient:
24async def get_async_client( 25 repository: str | URL, 26 refresh: bool = False, 27 limiter: Limiter | None = None, 28 config: Config | None = None, 29) -> AsyncRepositoryClient: 30 """ 31 Get an asynchronous client for the given repository. 32 33 :param repository: the repository alias or URL 34 :param refresh: whether to refresh the client configuration from the server 35 :param max_connections: the maximum number of parallel connections 36 :param config: the configuration to use. If not given, the configuration is loaded 37 from the configuration file. 38 :return: an asynchronous client for the repository 39 """ 40 if not config: 41 config = Config.from_file() 42 43 repository_config = config.find_repository(repository) 44 for async_client_class in async_client_classes(): 45 if await async_client_class.can_handle_repository(repository_config.url): 46 return await async_client_class.from_configuration( 47 repository_config, refresh=refresh, limiter=limiter 48 ) 49 raise ValueError(f"No async client found for repository {repository_config.url}")
Get an asynchronous client for the given repository.
Parameters
- repository: the repository alias or URL
- refresh: whether to refresh the client configuration from the server
- max_connections: the maximum number of parallel connections
- config: the configuration to use. If not given, the configuration is loaded from the configuration file.
Returns
an asynchronous client for the repository
class
AsyncRepositoryClient(typing.Protocol):
198class AsyncRepositoryClient(Protocol): 199 """An abstract client for NRP repositories. 200 201 Usually, subclasses of this class are not instantiated directly in your code. 202 To get an instance of a repository, use the high-level method `get_async_client`: 203 204 ``` 205 my_client = await async_client(config?, url=url, refresh=False/True) 206 my_client = await async_client(config?, alias=alias, refresh=False/True) 207 ``` 208 209 and then use the instance. 210 """ 211 212 # region info endpoint 213 @classmethod 214 async def can_handle_repository(cls, url: URL | str, verify_tls: bool = True) -> URL | None: 215 """Return if this client can handle a repository that contains the passed URL. 216 217 This method can make an http request or use any means to check if the repository 218 at the URL can be handled by this client. 219 220 :param url: any url within the repository (root, record api, html, documentation 221 if running on the same host, ...) 222 :param verify_tls: whether to verify tls (should be switched on for production) 223 :return: API url of the server or None if this client can not handle the repository 224 """ 225 ... 226 227 @classmethod 228 async def from_configuration(cls, 229 config: RepositoryConfig, 230 refresh: bool = False, 231 limiter: Limiter | None = None, 232 extra_tokens: dict[URL, str] | None = None 233 ) -> Self: 234 """Create a client from the given configuration. 235 236 :param config: the configuration for the repository 237 :param refresh: refresh the configuration by calling get_repository_info 238 """ 239 ... 240 241 async def get_repository_info(self, refresh: bool = True) -> RepositoryInfo: 242 """Get information about the repository. 243 244 This call is cached inside the RepositoryConfig instance. 245 246 :param refresh: refresh the info from the server 247 """ 248 ... 249 250 # endregion 251 252 @property 253 def records(self) -> AsyncRecordsClient: 254 """Return client for accessing records.""" 255 ... 256 257 # region records endpoint 258 259 # endregion 260 261 # region files endpoint 262 263 264 265 # endregion 266 267 # region requests endpoint 268 269 # endregion
An abstract client for NRP repositories.
Usually, subclasses of this class are not instantiated directly in your code.
To get an instance of a repository, use the high-level method get_async_client:
my_client = await async_client(config?, url=url, refresh=False/True)
my_client = await async_client(config?, alias=alias, refresh=False/True)
and then use the instance.
AsyncRepositoryClient(*args, **kwargs)
1767def _no_init_or_replace_init(self, *args, **kwargs): 1768 cls = type(self) 1769 1770 if cls._is_protocol: 1771 raise TypeError('Protocols cannot be instantiated') 1772 1773 # Already using a custom `__init__`. No need to calculate correct 1774 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1775 if cls.__init__ is not _no_init_or_replace_init: 1776 return 1777 1778 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1779 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1780 # searches for a proper new `__init__` in the MRO. The new `__init__` 1781 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1782 # instantiation of the protocol subclass will thus use the new 1783 # `__init__` and no longer call `_no_init_or_replace_init`. 1784 for base in cls.__mro__: 1785 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1786 if init is not _no_init_or_replace_init: 1787 cls.__init__ = init 1788 break 1789 else: 1790 # should not happen 1791 cls.__init__ = object.__init__ 1792 1793 cls.__init__(self, *args, **kwargs)
@classmethod
async def
can_handle_repository(cls, url: yarl.URL | str, verify_tls: bool = True) -> yarl.URL | None:
213 @classmethod 214 async def can_handle_repository(cls, url: URL | str, verify_tls: bool = True) -> URL | None: 215 """Return if this client can handle a repository that contains the passed URL. 216 217 This method can make an http request or use any means to check if the repository 218 at the URL can be handled by this client. 219 220 :param url: any url within the repository (root, record api, html, documentation 221 if running on the same host, ...) 222 :param verify_tls: whether to verify tls (should be switched on for production) 223 :return: API url of the server or None if this client can not handle the repository 224 """ 225 ...
Return if this client can handle a repository that contains the passed URL.
This method can make an http request or use any means to check if the repository at the URL can be handled by this client.
Parameters
- url: any url within the repository (root, record api, html, documentation if running on the same host, ...)
- verify_tls: whether to verify tls (should be switched on for production)
Returns
API url of the server or None if this client can not handle the repository
@classmethod
async def
from_configuration( cls, config: nrp_cmd.config.RepositoryConfig, refresh: bool = False, limiter: nrp_cmd.async_client.connection.Limiter | None = None, extra_tokens: dict[yarl.URL, str] | None = None) -> Self:
227 @classmethod 228 async def from_configuration(cls, 229 config: RepositoryConfig, 230 refresh: bool = False, 231 limiter: Limiter | None = None, 232 extra_tokens: dict[URL, str] | None = None 233 ) -> Self: 234 """Create a client from the given configuration. 235 236 :param config: the configuration for the repository 237 :param refresh: refresh the configuration by calling get_repository_info 238 """ 239 ...
Create a client from the given configuration.
Parameters
- config: the configuration for the repository
- refresh: refresh the configuration by calling get_repository_info
241 async def get_repository_info(self, refresh: bool = True) -> RepositoryInfo: 242 """Get information about the repository. 243 244 This call is cached inside the RepositoryConfig instance. 245 246 :param refresh: refresh the info from the server 247 """ 248 ...
Get information about the repository.
This call is cached inside the RepositoryConfig instance.
Parameters
- refresh: refresh the info from the server