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

AbstractRepoFetcher

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)#

Bases: abc.ABC

Abstract class to model stateful fetchers

Initialization

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

This is the method to be implemented by the stateful fetcher

abstract property description: str#

Description of this URI handler

abstract classmethod GetSchemeHandlers() → Mapping[str, DocumentedStatefulProtocolFetcher]#
abstract 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

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.AbstractRepoFetcher(progs: wfexs_backend.common.ProgsMapping = dict(), setup_block: Mapping[str, Any] | None = None)#

Bases: wfexs_backend.fetchers.AbstractStatefulFetcher

PRIORITY: ClassVar[int] = None#
abstract materialize_repo(repoURL: wfexs_backend.common.RepoURL, repoTag: RepoTag | None = None, repo_tag_destdir: PathLikePath | None = None, base_repo_destdir: PathLikePath | None = None, doUpdate: bool | None = True) → Tuple[pathlib.Path, RemoteRepo, Sequence[URIWithMetadata]]#
abstract 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, it should return None

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

Bases: wfexs_backend.fetchers.AbstractStatefulFetcher

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

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

Contents