From 8cce20d2e7a1465f81b5de06480c217566dab7cf Mon Sep 17 00:00:00 2001 From: MattWellie Date: Fri, 26 Jun 2026 10:20:09 +1000 Subject: [PATCH 1/8] adds a candidate metamist registration script --- cpg_utils/metamist_registration.py | 148 +++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100755 cpg_utils/metamist_registration.py diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py new file mode 100755 index 0000000..bbb1e09 --- /dev/null +++ b/cpg_utils/metamist_registration.py @@ -0,0 +1,148 @@ +#!/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 + +This will produce analysis records where: + - the analysis record is of a primary type defined by --type + - the analysis record has an 'output' value, defined by --output +""" + +import argparse +import json +import sys + +from collections import defaultdict +from typing import TypeAlias + +from metamist import graphql + + +RecursiveDict: TypeAlias = dict[str, 'str | RecursiveDict'] + + +def create_output_block(primary: str, type: str, secondary: list[str]) -> RecursiveDict: + """Populates the output dict based on the provided arguments.""" + + # create the outputs dictionary + outputs: RecursiveDict = { + type: { + 'basename': primary, + }, + } + + # take the list of key=value strings, snap'em, and add each to the secondary files list + if secondary: + secondary_kv = defaultdict(dict) + for keyvaluepair in secondary: + key, value = keyvaluepair.split('=') + secondary_kv[key] = { + 'basename': value, + } + + outputs[type]['secondary_files'] = secondary_kv + + return outputs + + +def 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).') + parser.add_argument('--sgs', nargs='+', help='Metamist Sequencing Group ID(s).') + parser.add_argument('--dry', action='store_true', help='Dry run, print only.') + args = parser.parse_args() + + if args.cohorts and args.sgs: + raise Exception('Cannot specify both --cohorts and --sgs') + + if not (args.cohorts or args.sgs): + raise Exception('Must specify either --cohorts or --sgs') + + query = """ + mutation updateAnalysis($project: String!, $analysis:AnalysisInput!) { + analysis { + createAnalysis(project:$project, analysis:$analysis) { + id + type + status + output + active + } + } + } + """ + + outputs = create_output_block( + primary=args.output, + type=args.type, + secondary=args.secondary, + ) + + variables = { + 'project': args.project, + 'analysis': { + 'type': args.type, + 'status': 'COMPLETED', + 'output': args.output, + 'outputs': outputs, + 'meta': {} + } + } + + # allow arbitrary values to be passed into meta. Not sure here how best to tolerate numerical values, if we have 'em + if args.meta: + meta_kv: dict[str, str] = {} + for keyvaluepair in args.meta: + key, value = keyvaluepair.split('=') + meta_kv[key] = value + variables['analysis']['meta'] = meta_kv + + if args.cohorts: + variables['analysis']['cohortIds'] = args.cohorts + if args.sgs: + variables['analysis']['sequencingGroupIds'] = args.sgs + + print(json.dumps(variables, indent=2)) + + if args.dry: + sys.exit(0) + + try: + result = graphql.query(query, variables) + print(f'Successfully recorded analysis to Metamist: {result}') + except Exception as e: + print(f'Error executing GraphQL query: {e}', file=sys.stderr) + sys.exit(1) + +if __name__ == '__main__': + main() From 7e07c5c04a64b2115591901f684b5f2f2ee58e12 Mon Sep 17 00:00:00 2001 From: MattWellie Date: Fri, 26 Jun 2026 11:09:44 +1000 Subject: [PATCH 2/8] and a patch - output is deprecated --- cpg_utils/metamist_registration.py | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py index bbb1e09..b87b71e 100755 --- a/cpg_utils/metamist_registration.py +++ b/cpg_utils/metamist_registration.py @@ -37,7 +37,6 @@ import json import sys -from collections import defaultdict from typing import TypeAlias from metamist import graphql @@ -49,23 +48,21 @@ def create_output_block(primary: str, type: str, secondary: list[str]) -> RecursiveDict: """Populates the output dict based on the provided arguments.""" - # create the outputs dictionary + # create the outputs dictionary, with the primary file's full path at the root outputs: RecursiveDict = { - type: { - 'basename': primary, - }, + 'basename': primary, } # take the list of key=value strings, snap'em, and add each to the secondary files list if secondary: - secondary_kv = defaultdict(dict) + secondary_kv: RecursiveDict = {} for keyvaluepair in secondary: key, value = keyvaluepair.split('=') secondary_kv[key] = { 'basename': value, } - outputs[type]['secondary_files'] = secondary_kv + outputs['secondary_files'] = secondary_kv return outputs @@ -95,7 +92,7 @@ def main(): id type status - output + outputs active } } @@ -113,7 +110,6 @@ def main(): 'analysis': { 'type': args.type, 'status': 'COMPLETED', - 'output': args.output, 'outputs': outputs, 'meta': {} } From 9951f2710a2f64f3e92dc6b4461a310c3538410b Mon Sep 17 00:00:00 2001 From: MattWellie Date: Fri, 26 Jun 2026 12:43:55 +1000 Subject: [PATCH 3/8] maybe work as CLI and imported method --- cpg_utils/metamist_registration.py | 234 ++++++++++++++++++++--------- 1 file changed, 164 insertions(+), 70 deletions(-) diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py index b87b71e..4d6d422 100755 --- a/cpg_utils/metamist_registration.py +++ b/cpg_utils/metamist_registration.py @@ -28,117 +28,211 @@ --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 'output' value, defined by --output + - 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 +import argparse # noqa: I001 import json import sys -from typing import TypeAlias +from metamist import exceptions, graphql + -from metamist import graphql +UPDATE_QUERY = """ + mutation updateAnalysis($project: String!, $analysis:AnalysisInput!) { + analysis { + createAnalysis(project:$project, analysis:$analysis) { + id + type + status + outputs + active + } + } + } + """ -RecursiveDict: TypeAlias = dict[str, 'str | RecursiveDict'] +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 -def create_output_block(primary: str, type: str, secondary: list[str]) -> RecursiveDict: + 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 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: RecursiveDict = { - 'basename': primary, - } + outputs: dict = {'basename': primary} - # take the list of key=value strings, snap'em, and add each to the secondary files list + # take the dict of type: file secondary files, and add to the analysis if secondary: - secondary_kv: RecursiveDict = {} - for keyvaluepair in secondary: - key, value = keyvaluepair.split('=') - secondary_kv[key] = { - 'basename': value, - } - - outputs['secondary_files'] = secondary_kv + outputs['secondary_files'] = { + key: {'basename': value} for key, value in secondary.items() + } return outputs -def main(): +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).') - parser.add_argument('--sgs', nargs='+', help='Metamist Sequencing Group ID(s).') - parser.add_argument('--dry', action='store_true', help='Dry run, print only.') + 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() - if args.cohorts and args.sgs: - raise Exception('Cannot specify both --cohorts and --sgs') + # 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, + ) - if not (args.cohorts or args.sgs): - raise Exception('Must specify either --cohorts or --sgs') - query = """ - mutation updateAnalysis($project: String!, $analysis:AnalysisInput!) { - analysis { - createAnalysis(project:$project, analysis:$analysis) { - id - type - status - outputs - active - } - } - } +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 Exception('Cannot specify both --cohorts and --sgs') + + if not (cohorts or sgs): + raise Exception('Must specify either --cohorts or --sgs') outputs = create_output_block( - primary=args.output, - type=args.type, - secondary=args.secondary, + primary=output, + secondary=secondary or {}, ) - variables = { - 'project': args.project, + variables: dict = { + 'project': project, 'analysis': { - 'type': args.type, + 'type': analysis_type, 'status': 'COMPLETED', 'outputs': outputs, - 'meta': {} - } + 'meta': meta, + 'cohortIds': cohorts or None, + 'sequencingGroupIds': sgs or None, + }, } - # allow arbitrary values to be passed into meta. Not sure here how best to tolerate numerical values, if we have 'em - if args.meta: - meta_kv: dict[str, str] = {} - for keyvaluepair in args.meta: - key, value = keyvaluepair.split('=') - meta_kv[key] = value - variables['analysis']['meta'] = meta_kv - - if args.cohorts: - variables['analysis']['cohortIds'] = args.cohorts - if args.sgs: - variables['analysis']['sequencingGroupIds'] = args.sgs - - print(json.dumps(variables, indent=2)) - - if args.dry: + if dry: + print('DRY RUN, would have POSTed the following analysis data:') + print(json.dumps(variables, indent=2)) sys.exit(0) try: - result = graphql.query(query, variables) + result = graphql.query(UPDATE_QUERY, variables) print(f'Successfully recorded analysis to Metamist: {result}') - except Exception as e: + except exceptions.ApiException as e: print(f'Error executing GraphQL query: {e}', file=sys.stderr) sys.exit(1) + if __name__ == '__main__': - main() + cli_main() From ee261dd4e6abfa2b5141333f6f8031b02109fdb7 Mon Sep 17 00:00:00 2001 From: MattWellie Date: Mon, 29 Jun 2026 11:09:11 +1000 Subject: [PATCH 4/8] remove hard failure --- cpg_utils/metamist_registration.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py index 4d6d422..e71ad1f 100755 --- a/cpg_utils/metamist_registration.py +++ b/cpg_utils/metamist_registration.py @@ -224,14 +224,13 @@ def create_new( if dry: print('DRY RUN, would have POSTed the following analysis data:') print(json.dumps(variables, indent=2)) - sys.exit(0) - - 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) - sys.exit(1) + 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__': From 77c9e73a71a44424d08be59dd9e03f748e14376b Mon Sep 17 00:00:00 2001 From: MattWellie Date: Mon, 29 Jun 2026 11:12:06 +1000 Subject: [PATCH 5/8] =?UTF-8?q?Bump=20version:=205.6.1=20=E2=86=92=205.7.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .bumpversion.cfg | 2 +- .github/workflows/docker.yaml | 2 +- setup.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) 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/setup.py b/setup.py index 4eacdc0..642cc34 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', From 6b0b9c18a1c8d5ef32f0caafe8f0d33ec1267dc7 Mon Sep 17 00:00:00 2001 From: MattWellie Date: Mon, 29 Jun 2026 11:15:13 +1000 Subject: [PATCH 6/8] make metamist dependency explicit --- setup.py | 1 + 1 file changed, 1 insertion(+) diff --git a/setup.py b/setup.py index 642cc34..5a755f7 100755 --- a/setup.py +++ b/setup.py @@ -27,6 +27,7 @@ 'google-auth-oauthlib', 'google-cloud-artifact-registry', 'google-cloud-secret-manager', + 'metamist', 'requests', 'tabulate', 'toml', From 53f85335c16d33f1db58f9599e7fa3b0d2522dde Mon Sep 17 00:00:00 2001 From: MattWellie Date: Tue, 30 Jun 2026 14:32:00 +1000 Subject: [PATCH 7/8] add existence checking --- cpg_utils/existence_checks.py | 92 ++++++++++++++++++++++++++++++ cpg_utils/metamist_registration.py | 26 ++++++++- 2 files changed, 116 insertions(+), 2 deletions(-) create mode 100644 cpg_utils/existence_checks.py 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 index e71ad1f..914b042 100755 --- a/cpg_utils/metamist_registration.py +++ b/cpg_utils/metamist_registration.py @@ -50,6 +50,7 @@ import json import sys +from cpg_utils.existence_checks import exists from metamist import exceptions, graphql @@ -91,6 +92,17 @@ def parse_cli_kv(input_kv: list[str]) -> dict[str, str]: 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() + if not exists(primary): + missing_files.add(primary) + for filepath in secondary.values(): + if not exists(filepath): + missing_files.add(filepath) + return missing_files + + def create_output_block( primary: str, secondary: dict[str, str], @@ -199,10 +211,20 @@ def create_new( """ # fail if the cohorts and sgs are both applied, or if neither is applied if cohorts and sgs: - raise Exception('Cannot specify both --cohorts and --sgs') + raise ValueError( + 'Cannot specify both --cohorts and --sgs CLI parameters for a single Analysis.', + ) if not (cohorts or sgs): - raise Exception('Must specify either --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.', + ) outputs = create_output_block( primary=output, From e5dcd2508f0ab3644f76b713ca8728df6f154c87 Mon Sep 17 00:00:00 2001 From: MattWellie Date: Tue, 30 Jun 2026 14:38:16 +1000 Subject: [PATCH 8/8] add gcs checking --- cpg_utils/metamist_registration.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/cpg_utils/metamist_registration.py b/cpg_utils/metamist_registration.py index 914b042..867ff9b 100755 --- a/cpg_utils/metamist_registration.py +++ b/cpg_utils/metamist_registration.py @@ -54,6 +54,7 @@ from metamist import exceptions, graphql +GS_PREFIX = 'gs://' UPDATE_QUERY = """ mutation updateAnalysis($project: String!, $analysis:AnalysisInput!) { analysis { @@ -95,14 +96,21 @@ def parse_cli_kv(input_kv: list[str]) -> dict[str, str]: 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() - if not exists(primary): - missing_files.add(primary) - for filepath in secondary.values(): + 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], @@ -225,6 +233,11 @@ def create_new( 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,