diff --git a/.bumpversion.cfg b/.bumpversion.cfg index 07131f2..8e77cee 100644 --- a/.bumpversion.cfg +++ b/.bumpversion.cfg @@ -1,5 +1,5 @@ [bumpversion] -current_version = 5.6.1 +current_version = 5.7.0 commit = True tag = False diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 5b89046..ae3fb62 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -5,7 +5,7 @@ on: - main env: - VERSION: 5.6.1 + VERSION: 5.7.0 permissions: contents: read diff --git a/cpg_utils/existence_checks.py b/cpg_utils/existence_checks.py new file mode 100644 index 0000000..6ace83b --- /dev/null +++ b/cpg_utils/existence_checks.py @@ -0,0 +1,92 @@ +""" +cached existence checking logic, ported from cpg-flow +https://github.com/populationgenomics/cpg-flow/blob/main/src/cpg_flow/utils.py + +this collection of methods can be used to: +- detect if a single file or directory exists, without caching +- detect if a single file or directory exists, with caching +- cache existence checks across a whole directory for rapid checking of multiple adjacent files/directories +""" + +import traceback # noqa: I001 +from os.path import basename, dirname + +import logging +from functools import lru_cache + +from cpg_utils import to_path, Path + + +@lru_cache +def exists(path: Path | str, verbose: bool = True) -> bool: + """ + `exists_not_cached` that caches the result. + + The python code runtime happens entirely during the workflow construction, + without waiting for it to finish, so there is no expectation that the object + existence status would change during the runtime. This, this function uses + `@lru_cache` to make sure that object existence is checked only once. + """ + return exists_not_cached(path, verbose) + + +def exists_not_cached(path_or_string: Path | str, verbose: bool = True) -> bool: + """ + Check if the object by path exists, where the object can be: + * local file, + * local directory, + * cloud object, + * cloud or local *.mt, *.ht, or *.vds Hail data, in which case it will check + for the existence of a corresponding _SUCCESS object instead. + @param path: path to the file/directory/object/mt/ht + @param verbose: print on each check + @return: True if the object exists + """ + + path = to_path(path_or_string) + + if path.suffix in {'.mt', '.ht'}: + path /= '_SUCCESS' + if path.suffix == '.vds': + path /= 'variant_data/_SUCCESS' + + if verbose: + try: + res = check_exists_path(path) + + # a failure to detect the parent folder causes a crash + # instead stick to a core responsibility - existence = False + except FileNotFoundError as fnfe: + logging.error(f'Failed checking {path}') + logging.error(f'{fnfe}') + return False + except BaseException as be: + traceback.print_exc() + logging.error(f'Failed checking {path}') + raise be + exist_debug_statement = 'exists' if res else 'missing' + logging.debug(f'Checked {path} [{exist_debug_statement}]') + return res + + return check_exists_path(path) + + +def check_exists_path(test_path: Path) -> bool: + """ + Check whether a path exists using a cached per-directory listing. + NB. reversion to Strings prevents a get call, which is typically + forbidden to local users + """ + return basename(str(test_path)) in get_contents_of_path(dirname(str(test_path))) + + +@lru_cache +def get_contents_of_path(test_path: str) -> set[str]: + """ + Get the contents of a GCS path, returning non-complete paths, eg: + + > get_contents_of_path('gs://my-bucket/my-dir/') + {'my-file.txt'} + + """ + return {f.name for f in to_path(test_path.rstrip('/')).iterdir()} diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py new file mode 100755 index 0000000..867ff9b --- /dev/null +++ b/cpg_utils/metamist_registration.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 + +""" +Generic Metamist analysis generation script, must be executable + +- takes a path to a primary output, and optionally one or more secondary analyses + - due to the fixed dictionary structure (type of each secondary is within schema), this requires 'k=v k2=v2' args. +- requires an analysis type to register this as +- can take either SGs or Cohorts to associate the analysis entry with, but not both +- optionally can use the same k=v breaking to register metadata + +e.g. CLI + +# minimal example +python3 -m cpg_utils.metamist_registration \ + --project \ + --output \ + --type cram \ + --cohorts COH1234 \ + --dry + +# with optional metaadta +python3 -m cpg_utils.metamist_registration \ + --project \ + --output \ + --type cram \ + --secondary index=path/to/index html=/path/to/html \ + --meta stage=name sequencing_type=genome \ + --cohorts COH1234 COH5678 + +# imported equivalent using a method call +from cpg_utils.metamist_registration import create_new +create_new( + project=, + output=, + analysis_type=cram, + meta={'stage': 'name', 'sequencing_type': 'genome'}, + cohorts=['COH1234', 'COH5678'], +) + +This will produce analysis records where: + - the analysis record is of a primary type defined by --type + - the analysis record has an 'outputs' dictionary + - outputs.path is the value of --output + - if used, secondary files will be nested inside `analysis.outputs.secondary_files.TYPE.path` + - if used, analysis.meta will contain all the meta key=value pairs +""" + +import argparse # noqa: I001 +import json +import sys + +from cpg_utils.existence_checks import exists +from metamist import exceptions, graphql + + +GS_PREFIX = 'gs://' +UPDATE_QUERY = """ + mutation updateAnalysis($project: String!, $analysis:AnalysisInput!) { + analysis { + createAnalysis(project:$project, analysis:$analysis) { + id + type + status + outputs + active + } + } + } + """ + + +def parse_cli_kv(input_kv: list[str]) -> dict[str, str]: + """ + Takes a list of key-value pairs and parses them into a dictionary. + Used in populating both meta and secondary files. + + Args: + input_kv: list[str] a list of key-value pairs passed form the CLI + + Returns: + a dictionary of key-value pairs built from the original strings + """ + broken_kv = {} + for keyvaluepair in input_kv: + if '=' not in keyvaluepair: + raise ValueError(f'Found key=value entry which lacks a "=": {keyvaluepair}') + + key, value = keyvaluepair.split('=') + if key in broken_kv: + raise ValueError(f'Duplicate key provided: {key}') + broken_kv[key] = value + return broken_kv + + +def find_missing_files(primary: str, secondary: dict[str, str]) -> set[str]: + """For the primary and secondary files, detect if any are missing. Return all missing Paths.""" + missing_files: set[str] = set() + for filepath in [primary, *list(secondary.values())]: + if not exists(filepath): + missing_files.add(filepath) + return missing_files + + +def find_non_gcs_files(primary: str, secondary: dict[str, str]) -> set[str]: + """Checks for a `gs://` prefix on all files.""" + non_gs_files: set[str] = set() + for filepath in [primary, *list(secondary.values())]: + if not filepath.startswith('gs://'): + non_gs_files.add(filepath) + return non_gs_files + + +def create_output_block( + primary: str, + secondary: dict[str, str], +) -> dict: + """Populates the output dict based on the provided arguments.""" + + # create the outputs dictionary, with the primary file's full path at the root + outputs: dict = {'basename': primary} + + # take the dict of type: file secondary files, and add to the analysis + if secondary: + outputs['secondary_files'] = { + key: {'basename': value} for key, value in secondary.items() + } + + return outputs + + +def cli_main(): + parser = argparse.ArgumentParser(description='Record MultiQC results to Metamist.') + parser.add_argument( + '--project', + required=True, + help='The Metamist project name.', + ) + parser.add_argument( + '--output', + required=True, + help='Path of the primary output file/dir.', + ) + parser.add_argument( + '--type', + required=True, + help='Analysis type, must be from valid enum.', + ) + parser.add_argument( + '--secondary', + nargs='+', + help='Optional, list of k=v pairs for secondary analyses.', + default=[], + ) + parser.add_argument( + '--meta', + nargs='+', + help='List of k=v metadata pairs, optional.', + default=[], + ) + parser.add_argument( + '--cohorts', + nargs='+', + help='Metamist cohort ID(s).', + default=[], + ) + parser.add_argument( + '--sgs', + nargs='+', + help='Metamist Sequencing Group ID(s).', + default=[], + ) + parser.add_argument( + '--dry', + action='store_true', + help='Dry run, print only.', + ) + args = parser.parse_args() + + # do some transformation of the input meta/secondary file dictionaries + meta_dict = parse_cli_kv(args.meta) + secondary_dict = parse_cli_kv(args.secondary) + + # call the main analysis generation method + create_new( + project=args.project, + output=args.output, + analysis_type=args.type, + cohorts=args.cohorts, + sgs=args.sgs, + dry=args.dry, + meta=meta_dict, + secondary=secondary_dict, + ) + + +def create_new( + project: str, + output: str, + analysis_type: str, + meta: dict[str, str] | None = None, + secondary: dict[str, str] | None = None, + cohorts: list[str] | None = None, + sgs: list[str] | None = None, + dry: bool = False, +) -> None: + """ + main method, takes the provided inputs and creates a new analysis entry + + Args: + project: str, the name of the project to create the analysis entry in + output: str, the primary output path of the analysis entry + analysis_type: str, the analysis type (must be from valid enum in metamist) + meta: optional dict, if provided this will be added as the analysis.meta dictionary + secondary: optional dict, if provided each element of this {type: string} dict will be added as a secondary file + cohorts: optional list, COHort IDs to attribute the analysis to. Mutually exclusive with sgs + sgs: optional list, Sequencing Group IDs to attribute the analysis to. Mutually exclusive with cohorts + dry: bool, if True, payload is printed, but not sent + """ + # fail if the cohorts and sgs are both applied, or if neither is applied + if cohorts and sgs: + raise ValueError( + 'Cannot specify both --cohorts and --sgs CLI parameters for a single Analysis.', + ) + + if not (cohorts or sgs): + raise ValueError( + 'You must specify either --cohorts or --sgs for a single Analysis object.', + ) + + if missing_files := find_missing_files(primary=output, secondary=secondary or {}): + missing_file_string = ', '.join(sorted(missing_files)) + raise ValueError( + f'Missing files detected: {missing_file_string}.\nThis can only be used for extant files.', + ) + if non_gcs_files := find_non_gcs_files(primary=output, secondary=secondary or {}): + missing_file_string = ', '.join(sorted(non_gcs_files)) + raise ValueError( + f'Non-GCS files detected: {missing_file_string}.\nThis can only be used for files stored in GCS.', + ) + + outputs = create_output_block( + primary=output, + secondary=secondary or {}, + ) + + variables: dict = { + 'project': project, + 'analysis': { + 'type': analysis_type, + 'status': 'COMPLETED', + 'outputs': outputs, + 'meta': meta, + 'cohortIds': cohorts or None, + 'sequencingGroupIds': sgs or None, + }, + } + + if dry: + print('DRY RUN, would have POSTed the following analysis data:') + print(json.dumps(variables, indent=2)) + else: + try: + result = graphql.query(UPDATE_QUERY, variables) + print(f'Successfully recorded analysis to Metamist: {result}') + except exceptions.ApiException as e: + print(f'Error executing GraphQL query: {e}', file=sys.stderr) + raise e + + +if __name__ == '__main__': + cli_main() diff --git a/setup.py b/setup.py index 4eacdc0..5a755f7 100755 --- a/setup.py +++ b/setup.py @@ -8,7 +8,7 @@ setup( name='cpg-utils', # This tag is automatically updated by bumpversion - version='5.6.1', + version='5.7.0', description='Library of convenience functions specific to the CPG', long_description=long_description, long_description_content_type='text/markdown', @@ -27,6 +27,7 @@ 'google-auth-oauthlib', 'google-cloud-artifact-registry', 'google-cloud-secret-manager', + 'metamist', 'requests', 'tabulate', 'toml',