Source code for cached_path.schemes

from typing import Set, Type

from .gs import GsClient
from .hf import hf_get_from_cache
from .http import HttpClient
from .s3 import S3Client
from .scheme_client import SchemeClient

__all__ = ["GsClient", "HttpClient", "S3Client", "SchemeClient", "hf_get_from_cache"]

try:
    from .beaker import BeakerClient

    __all__.append("BeakerClient")
except (ImportError, ModuleNotFoundError):
    BeakerClient = None  # type: ignore


_SCHEME_TO_CLIENT = {}


[docs]def add_scheme_client(client: Type[SchemeClient]) -> None: """ Add a new :class:`SchemeClient`. This can be used to extend :func:`cached_path.cached_path()` to handle custom schemes, or handle existing schemes differently. """ global _SCHEME_TO_CLIENT if isinstance(client.scheme, tuple): for scheme in client.scheme: _SCHEME_TO_CLIENT[scheme] = client elif isinstance(client.scheme, str): _SCHEME_TO_CLIENT[client.scheme] = client else: raise ValueError(f"Unexpected type for {client} scheme: {client.scheme}")
for client in (HttpClient, S3Client, GsClient): add_scheme_client(client) # type: ignore if BeakerClient is not None: add_scheme_client(BeakerClient) def get_scheme_client(resource: str) -> SchemeClient: """ Get the right client for the given resource. """ maybe_scheme = resource.split("://")[0] return _SCHEME_TO_CLIENT.get(maybe_scheme, HttpClient)(resource) def get_supported_schemes() -> Set[str]: """ Return all supported URL schemes. """ return set(_SCHEME_TO_CLIENT.keys()) | {"hf"}