nf_core.utils

Common utility functions for the nf-core python package.

exceptionnf_core.utils.CommandError(cmd: str, returncode: int, stdout: bytes, stderr: bytes)

Bases: RuntimeError

Raised by run_cmd when a command exits with a non-zero code.

Keeps the captured streams, so callers can inspect the command’s output instead of parsing the error message. output holds both streams decoded, since Nextflow prints error details to stdout and only a summary to stderr.

classnf_core.utils.ContainerRegistryUrls(*values)

Bases: Enum

GALAXY_SINGULARITY= ‘depot.galaxyproject.org/singularity’

SEQERA_DOCKER= ‘community.wave.seqera.io/library’

SEQERA_SINGULARITY= ‘community-cr-prod.seqera.io/docker/registry/v2’

classnf_core.utils.Pipeline(wf_path: Path)

Bases: object

Object to hold information about a local pipeline.

• Parameters: path (str) – The path to the nf-core pipeline directory.

conda_config

The parsed conda configuration file content (environment.yml).

• Type: dict

conda_package_info

The conda package(s) information, based on the API requests to Anaconda cloud.

• Type: dict

nf_config

The Nextflow pipeline configuration file content.

• Type: dict

files

A list of files found during the linting process.

• Type: list

git_sha

The git sha for the repo commit / current GitHub pull-request ($GITHUB_PR_COMMIT)

• Type: str

minNextflowVersion

The minimum required Nextflow version to run the pipeline.

• Type: str

wf_path

Path to the pipeline directory.

• Type: str

pipeline_name

The pipeline name, without the nf-core tag, for example hlatyping.

• Type: str

schema_obj

A PipelineSchema object

• Type: obj

_fp(fn: str | Path) → Path

Convenience function to get full path to a file in the pipeline

_load() → bool

Run core load functions

_load_conda_environment() → bool

Try to load the pipeline environment.yml file, if it exists

conda_config: dict

conda_package_info: dict

files: list[Path]

git_sha: str | None

list_files() → list[Path]

Get a list of all files in the pipeline

load_pipeline_config() → bool

Get the nextflow config for this pipeline

Once loaded, set a few convenience reference class attributes

minNextflowVersion: str | None

nf_config: dict

pipeline_name: str | None

pipeline_prefix: str | None

repo: git.Repo | None

schema_obj: PipelineSchema | None

nf_core.utils._load_cached_remote_version() → str | None

nf_core.utils._load_version_cache() → dict

nf_core.utils._nfcore_question_style()

nf_core.utils._spawn_remote_version_refresh(source_url: str) → None

Kick off a detached background process to refresh the remote version cache.

nf_core.utils.anaconda_package(dep, dep_channels=None)

Query conda package information.

Sends a HTTP GET request to the Anaconda remote API.

• Parameters:
  • dep (str) – A conda package name.
  • dep_channels (list) – list of conda channels to use
• Raises:
  • A LookupError**,** if the connection fails or times out or gives an unexpected status code –
  • A ValueError**,** if the package name can not be found (404) –

nf_core.utils.check_if_outdated(current_version=None, remote_version=None, source_url='https://nf-co.re/tools_version')

Check if the current version of nf-core is outdated

nf_core.utils.check_nextflow_version(minimal_nf_version: tuple[int, int, int, bool], silent=False) → bool

Check the version of Nextflow installed on the system.

• Parameters:
  • minimal_nf_version (tuple *[*int , int , int , bool ]) – The minimal version of Nextflow required.
  • silent (bool) – Whether to log the version or not.
• Returns: True if the installed version is greater than or equal to minimal_nf_version
• Return type: bool

nf_core.utils.custom_yaml_dumper()

Overwrite default PyYAML output to make Prettier YAML linting happy

nf_core.utils.determine_base_dir(directory: Path | str = '.') → Path

nf_core.utils.fetch_wf_config(wf_path: Path, cache_config: bool = True) → dict

Uses Nextflow to retrieve the the configuration variables from a Nextflow workflow.

• Parameters:
  • wf_path (str) – Nextflow workflow file system path.
  • cache_config (bool) – cache configuration or not (def. True)
• Returns: Workflow configuration settings.
• Return type: dict

nf_core.utils.file_md5(fname)

Calculates the md5sum for a file on the disk.

• Parameters: fname (str) – Path to a local file.

nf_core.utils.get_biocontainer_tag(package, version)

Given a bioconda package and version, looks for Docker and Singularity containers using the biocontaineres API, e.g.: https://api.biocontainers.pro/ga4gh/trs/v2/tools/{tool}/versions/{tool}-{version} Returns the most recent container versions by default. :param package: A bioconda package name. :type package: str :param version: Version of the bioconda package :type version: str

• Raises:
  • A LookupError**,** if the connection fails or times out or gives an unexpected status code –
  • A ValueError**,** if the package name can not be found (404) –

nf_core.utils.get_first_available_path(directory: Path | str, paths: list[str]) → Path | None

nf_core.utils.get_nf_version() → tuple[int, int, int, bool] | None

Get the version of Nextflow installed on the system. Cached for the lifetime of the process.

nf_core.utils.get_repo_commit(pipeline, commit_id)

Check if the repo contains the requested commit_id, and expand it to long form if necessary.

• Parameters:
  • pipeline (str) – GitHub repo username/repo
  • commit_id – The requested commit ID (SHA). It can be in standard long/short form, or any length.
• Returns: String or None
• Return type: commit_id

nf_core.utils.get_repo_releases_branches(pipeline, wfs)

Fetches details of a nf-core workflow to download.

• Parameters:
  • pipeline (str) – GitHub repo username/repo
  • wfs – A nf_core.pipelines.list.Workflows() object, where get_remote_workflows() has been called.
• Returns: Array of releases, Array of branches
• Return type: wf_releases, wf_branches (tuple)
• Raises: LockupError**,** if the pipeline can not be found. –

nf_core.utils.get_wf_files(wf_path: Path)

Return a list of all files in a directory (ignores .gitigore files)

nf_core.utils.is_file_binary(path)

Check file path to see if it is a binary file

nf_core.utils.is_interactive() → bool

Check if the current session is interactive (has a TTY on stdin, stdout, and stderr).

nf_core.utils.is_pipeline_directory(wf_path)

Checks if the specified directory have the minimum required files (‘main.nf’, ‘nextflow.config’) for a pipeline directory

• Parameters: wf_path (str) – The directory to be inspected
• Raises: UserWarning – If one of the files are missing

nf_core.utils.is_relative_to(path1, path2)

Checks if a path is relative to another.

Should mimic Path.is_relative_to which not available in Python < 3.9

path1 (Path | str): The path that could be a subpath path2 (Path | str): The path the could be the superpath

nf_core.utils.lazy_attrs(module_globals: dict, mapping: dict[str, str | Callable])

Build module-level __getattr__ and __dir__ for lazy attributes (PEP 562).

Each attribute maps either to the dotted path of the module to import it from, or to a zero-argument callable that builds it. Resolved attributes are cached in the module’s globals, so the hook only fires once per name.

Usage:

__getattr__, __dir__ = lazy_attrs(globals(), {"PipelineCreateApp": "nf_core.pipelines.create"})

nf_core.utils.load_tools_config(directory: str | Path = '.') → tuple[Path | None,NFCoreYamlConfig| None]

Parse the nf-core.yml configuration file

Look for a file called either .nf-core.yml or .nf-core.yaml

Also looks for the deprecated file .nf-core-lint.yml/yaml and issues a warning that this file will be deprecated in the future

Returns the loaded config dict or False, if the file couldn’t be loaded

nf_core.utils.nested_delitem(d, keys)

Deletes a key from a nested dictionary

• Parameters:
  • d (dict) – the nested dictionary to traverse
  • keys (list *[*Any ]) – A list of keys to iteratively traverse, deleting the final one

nf_core.utils.nested_setitem(d, keys, value)

Sets the value in a nested dict using a list of keys to traverse

• Parameters:
  • d (dict) – the nested dictionary to traverse
  • keys (list *[*Any ]) – A list of keys to iteratively traverse
  • value (Any) – The value to be set for the last key in the chain

nf_core.utils.parse_anaconda_licence(anaconda_response, version=None)

Given a response from the anaconda API using anaconda_package, parse the software licences.

Returns: Set of licence types

nf_core.utils.parse_nextflow_json(text: str) → Any

Parse the JSON output of a Nextflow command, ignoring status lines printed around it.

Nextflow can print status lines to stdout next to the JSON, e.g. [PIPELINE] … or, before 26.05, the Seqera Platform monitoring URL.

• Raises: json.JSONDecodeError – If the text contains no valid JSON object.

nf_core.utils.pip_package(dep)

Query PyPI package information.

Sends a HTTP GET request to the PyPI remote API.

• Parameters: dep (str) – A PyPI package name.
• Raises:
  • A LookupError**,** if the connection fails or times out –
  • A ValueError**,** if the package name can not be found –

nf_core.utils.plural_es(list_or_int)

Return a ‘es’ if the input is not one or has not the length of one.

nf_core.utils.plural_s(list_or_int)

Return an s if the input is not one or has not the length of one.

nf_core.utils.plural_y(list_or_int)

Return ‘ies’ if the input is not one or has not the length of one, else ‘y’.

nf_core.utils.poll_nfcore_web_api(api_url: str, post_data: dict | None = None) → dict

Poll the nf-core website API

Takes argument api_url for URL

Expects API response to be valid JSON and contain a top-level ‘status’ key.

nf_core.utils.pretty_nf_version(version: tuple[int, int, int, bool]) → str

nf_core.utils.prompt_pipeline_release_branch(wf_releases: list[dict[str, Any]], wf_branches: dict[str, Any], multiple: bool = False) → tuple[Any, list[str]]

Prompt for pipeline release / branch

• Parameters:
  • wf_releases (array) – Array of repo releases as returned by the GitHub API
  • wf_branches (array) – Array of repo branches, as returned by the GitHub API
  • multiple (bool) – Allow selection of multiple releases & branches (for Seqera Platform)
• Returns: Selected release / branch or False if no releases / branches available
• Return type: choice (questionary.Choice or bool)

nf_core.utils.prompt_remote_pipeline_name(wfs)

Prompt for the pipeline name with questionary

• Parameters: wfs – A nf_core.pipelines.list.Workflows() object, where get_remote_workflows() has been called.
• Returns: GitHub repo - username/repo
• Return type: pipeline (str)
• Raises: AssertionError**,** if pipeline cannot be found –

nf_core.utils.read_module_name(main_nf: Path) → str | None

Return the process name declared in a Nextflow main.nf file, or None.

nf_core.utils.rich_force_colors()

Check if any environment variables are set to force Rich to use coloured output

nf_core.utils.run_cmd(executable: str, cmd: str) → tuple[bytes, bytes] | None

Run a specified command and capture the output. Handle errors nicely.

nf_core.utils.set_wd(path: Path) → Generator[None, None, None]

Sets the working directory for this context.

• Parameters: path (Path) – Path to the working directory to be used inside this context.

nf_core.utils.set_wd_tempdir(base_dir: Path | None = None) → Generator[Path, None, None]

Context manager to create a tempdir and change into it, ensuring its removal and a return to the original working directory on exit (including exceptions).

• Parameters: base_dir – Directory in which to create the tempdir. Defaults to the system temp location.

nf_core.utils.setup_nfcore_cachedir(cache_fn: str | Path | None = None) → Path

Sets up local caching for caching files between sessions.

Creates and returns a subdirectory of the nf-core cache directory, or the cache directory itself if no subdirectory is given.

nf_core.utils.setup_requests_cachedir() → dict[str, Path | timedelta | str]

Sets up local caching for faster remote HTTP requests.

Caching directory will be set up in the user’s home directory under a .config/nf-core/cache_* subdir.

Uses requests_cache monkey patching. Also returns the config dict so that we can use the same setup with a Session.

nf_core.utils.sort_dictionary(d: dict) → dict

Sorts a nested dictionary recursively

nf_core.utils.strip_ansi_codes(string, replace_with='')

Strip ANSI colouring codes from a string to return plain text.

From Stack Overflow: https://stackoverflow.com/a/14693789/713980

nf_core.utils.unquote(s: str) → str

Remove paired quotes (single or double) from start and end of string.

Uses ast.literal_eval to safely parse Python string literals, preserving the original string if it’s not a valid literal.

Special handling for ruamel.yaml DoubleQuotedScalarString to preserve strings that should not be converted to numbers (e.g., “123” stays as string).

• Parameters: s – String potentially containing quotes
• Returns: String with outer quotes removed if present, otherwise original string

nf_core.utils.validate_file_md5(file_name, expected_md5hex)

Validates the md5 checksum of a file on disk.

• Parameters:
  • file_name (str) – Path to a local file.
  • expected (str) – The expected md5sum.
• Raises: IOError**,** if the md5sum does not match the remote sum. –

nf_core.utils.wait_cli_function(poll_func: Callable[[], bool], refresh_per_second: int = 20) → None

Display a command-line spinner while calling a function repeatedly.

Keep waiting until that function returns True

• Parameters:
  • poll_func (function) – Function to call
  • refresh_per_second (int) – Refresh this many times per second. Default: 20.
• Returns: None. Just sits in an infinite loop until the function returns True.