wfexs_backend.fetchers

Contents

wfexs_backend.fetchers#

Subpackages#

Submodules#

Package Contents#

Classes#

ProtocolFetcherReturn

DocumentedProtocolFetcher

DocumentedStatefulProtocolFetcher

AbstractStatefulFetcher

Abstract class to model stateful fetchers

RepoType

RepoGuessFlavor

RemoteRepo

Remote repository description

MaterializedRepo

AbstractSchemeRepoFetcher

AbstractStatefulStreamingFetcher

Data#

DEFAULT_PRIORITY

API#

class wfexs_backend.fetchers.ProtocolFetcherReturn#

Bases: typing.NamedTuple

kind_or_resolved: AnyURI | ContentKind | Sequence[AnyURI] = None#
metadata_array: Sequence[URIWithMetadata] = None#
licences: Tuple[URIType, ...] | None = None#
wfexs_backend.fetchers.DEFAULT_PRIORITY: Final[int] = 0#
class wfexs_backend.fetchers.DocumentedProtocolFetcher#

Bases: typing.NamedTuple

fetcher: wfexs_backend.fetchers.ProtocolFetcher = None#
description: str = None#
priority: int = None#
class wfexs_backend.fetchers.DocumentedStatefulProtocolFetcher#

Bases: typing.NamedTuple

fetcher_class: Type[AbstractStatefulFetcher] = None#
description: str | None = None#
priority: int = None#
exception wfexs_backend.fetchers.FetcherException(msg: str, code: str | int | None = None, reason: str | None = None)#

Bases: wfexs_backend.common.AbstractWfExSException

code: str | int | None = None#
reason: str | None = None#
exception wfexs_backend.fetchers.InvalidFetcherException(msg: str, code: str | int | None = None, reason: str | None = None)#

Bases: wfexs_backend.fetchers.FetcherException

exception wfexs_backend.fetchers.FetcherInstanceException(msg: str, code: str | int | None = None, reason: str | None = None)#

Bases: wfexs_backend.fetchers.FetcherException

class wfexs_backend.fetchers.AbstractStatefulFetcher(progs: wfexs_backend.common.ProgsMapping = dict(), setup_block: Mapping[str, Any] | None = None, scheme_catalog: SchemeCatalog | None = None)#

Bases: abc.ABC

Abstract class to model stateful fetchers

Initialization

PRIORITY: ClassVar[int] = None#
ENABLED: ClassVar[bool] = True#
abstractmethod fetch(remote_file: wfexs_backend.common.URIType, cachedFilename: wfexs_backend.common.PathLikePath, secContext: SecurityContextConfig | None = None, explicit_redirects: bool = False) → wfexs_backend.fetchers.ProtocolFetcherReturn#

This is the method to be implemented by the stateful fetcher

abstract property description: str#

Description of this URI handler

abstractmethod classmethod GetSchemeHandlers() → Mapping[str, DocumentedStatefulProtocolFetcher]#
abstractmethod classmethod GetNeededPrograms() → Sequence[SymbolicName]#
static ParseAndRemoveCredentials(remote_file: wfexs_backend.common.URIType) → Tuple[parse.ParseResult, URIType]#
exception wfexs_backend.fetchers.RepoGuessException(msg: str, code: str | int | None = None, reason: str | None = None)#

Bases: wfexs_backend.fetchers.FetcherException

exception wfexs_backend.fetchers.OfflineRepoGuessException(msg: str, code: str | int | None = None, reason: str | None = None)#

Bases: wfexs_backend.fetchers.RepoGuessException

class wfexs_backend.fetchers.RepoType(*args, **kwds)#

Bases: enum.Enum

Git = 'git'#
Raw = 'raw'#
Other = 'other'#
SoftwareHeritage = 'swh'#
TRS = 'trs'#
classmethod _undeprecate_table() → Mapping[str, str]#
class wfexs_backend.fetchers.RepoGuessFlavor(*args, **kwds)#

Bases: enum.Enum

GitHub = 'github'#
GitLab = 'gitlab'#
BitBucket = 'bitbucket'#
class wfexs_backend.fetchers.RemoteRepo#

Bases: typing.NamedTuple

Remote repository description

repo_url: wfexs_backend.common.RepoURL = None#
tag: RepoTag | None = None#
rel_path: RelPath | None = None#
repo_type: RepoType | None = None#
web_url: URIType | None = None#
guess_flavor: RepoGuessFlavor | None = None#
checkout: RepoTag | None = None#
gen_repo_desc() → RepoDesc | None#
get_checkout() → wfexs_backend.common.RepoTag#
class wfexs_backend.fetchers.MaterializedRepo#

Bases: typing.NamedTuple

local: pathlib.Path = None#
repo: wfexs_backend.fetchers.RemoteRepo = None#
metadata_array: Sequence[URIWithMetadata] = None#
upstream_repo: RemoteRepo | None = None#
recommends_upstream: bool = False#
class wfexs_backend.fetchers.AbstractSchemeRepoFetcher(scheme_catalog: wfexs_backend.scheme_catalog.SchemeCatalog, progs: wfexs_backend.common.ProgsMapping = dict(), setup_block: Mapping[str, Any] | None = None)#

Bases: wfexs_backend.fetchers.AbstractStatefulFetcher

PRIORITY: ClassVar[int] = None#

This abstract subclass is used to force the initialization of the scheme catalog instance

abstractmethod materialize_repo_from_repo(repo: wfexs_backend.fetchers.RemoteRepo, repo_tag_destdir: PathLikePath | None = None, base_repo_destdir: PathLikePath | None = None, doUpdate: bool | None = True) → wfexs_backend.fetchers.MaterializedRepo#

Subclasses have to implement this method, which is used to materialize a repository described by a RemoteRepo instance.

Parameters:

  • repo (class: wfexs_backend.fetchers.RemoteRepo) – The description of the repository to be materialized.

  • repo_tag_destdir (str, os.PathLike[str], optional) – Destination of the materialized repo.

  • base_repo_destdir (str, os.PathLike[str], optional) – If repo_tag_destdir is None, parent directory of the newly created destination directory for the repo.

  • doUpdate (bool) – Should the code try updating an already materialized repo? Defaults to False

The returned tuple has next elements: * The local path where the repo was materialized. * A RemoteRepo instance. * The metadata gathered through the materialisation process. * An optional, upstream URI representing the repo. For instance,

in the case of a TRS or a SWH hosted repo, the registered upstream URL.

abstractmethod build_pid_from_repo(remote_repo: wfexs_backend.fetchers.RemoteRepo) → str | None#

This method is required to generate a PID which usually represents an element (usually a workflow) in a repository. If the fetcher does not recognize the type of repo, either using repo_url content or the repo type in the worst case, it should return None

Calling this method in offline mode should be safe

abstractmethod classmethod GuessRepoParams(orig_wf_url: URIType | parse.ParseResult, logger: logging.Logger | None = None, fail_ok: bool = False, offline: bool = False) → RemoteRepo | None#
class wfexs_backend.fetchers.AbstractStatefulStreamingFetcher(progs: wfexs_backend.common.ProgsMapping = dict(), setup_block: Mapping[str, Any] | None = None, scheme_catalog: SchemeCatalog | None = None)#

Bases: wfexs_backend.fetchers.AbstractStatefulFetcher

fetch(remote_file: wfexs_backend.common.URIType, cachedFilename: wfexs_backend.common.PathLikePath, secContext: SecurityContextConfig | None = None, explicit_redirects: bool = False) → wfexs_backend.fetchers.ProtocolFetcherReturn#
abstractmethod streamfetch(remote_file: wfexs_backend.common.URIType, dest_stream: IO[bytes], secContext: SecurityContextConfig | None = None, explicit_redirects: bool = False) → wfexs_backend.fetchers.ProtocolFetcherReturn#

This is the method to be implemented by the stateful streaming fetcher which can receive as destination a byte stream

Contents