Data Search Command

wigwam/cli/build_commands.py

@@ -26,6 +26,7 @@ def init_build_parsers(subparsers: argparse._SubParsersAction) -> None:
         cmake-config,
         cmake-compile,
         cmake-install,
+        make-distrib,
     and more are being added.
 
     Parameters

wigwam/cli/cli.py

@@ -5,6 +5,7 @@
 from ..defaults import universal_tag_prefix
 from ._utils import help_formatter
 from .build_commands import build_command_names, init_build_parsers, run_build
+from .data_commands import init_data_parsers, run_data
 from .setup_commands import init_setup_parsers, run_setup
 from .util_commands import init_util_parsers, run_util
 
@@ -27,6 +28,7 @@ def initialize_parser() -> argparse.ArgumentParser:
 
     init_setup_parsers(subparsers, prefix)
     init_build_parsers(subparsers)
+    init_data_parsers(subparsers)
     init_util_parsers(subparsers, prefix)
 
     return parser
@@ -39,6 +41,8 @@ def main(args: Sequence[str] = sys.argv[1:]):
     del args_parsed.command
     if command == "setup":
         run_setup(args_parsed)
+    elif command == "data":
+        run_data(args_parsed)
     elif command in build_command_names():
         run_build(args_parsed, command)
     else:

wigwam/cli/data_commands.py

@@ -0,0 +1,76 @@
+import argparse
+from pathlib import Path
+
+from ..data_commands import data_search
+from ..defaults import default_workflowdata_path
+from ._utils import help_formatter
+
+
+def init_data_parsers(subparsers: argparse._SubParsersAction) -> None:
+    """
+    Augment an argument parser with setup commands.
+
+    Parameters
+    -------
+    parser : argparse.ArgumentParser
+        The parser to add setup commands to.
+    """
+    search_params = argparse.ArgumentParser(add_help=False)
+    search_params.add_argument(
+        "--data-file",
+        "-f",
+        type=Path,
+        default=default_workflowdata_path(),
+        metavar="FILENAME",
+        help="The filename of the repository metadata file.",
+    )
+    search_params.add_argument(
+        "--tags",
+        "-t",
+        nargs="+",
+        action="append",
+        default=[],
+        metavar="TAG",
+        help="A set of data repository tags. Can be used multiple times.",
+    )
+    search_params.add_argument(
+        "--names",
+        "-n",
+        nargs="+",
+        default=[],
+        metavar="NAME",
+        help="A set of data repository names.",
+    )
+    search_params.add_argument(
+        "--all",
+        "-a",
+        action="store_true",
+        default=False,
+        help="If used, get all repositories. Other search parameters will be ignored.",
+    )
+
+    data_parser = subparsers.add_parser(
+        "data", help="Perform data operations.", formatter_class=help_formatter
+    )
+    data_subparsers = data_parser.add_subparsers(dest="data_subcommand")
+
+    search_parser = data_subparsers.add_parser(
+        "search",
+        parents=[search_params],
+        help="Search a file for repository metadata.",
+        formatter_class=help_formatter,
+    )
+    search_parser.add_argument(
+        "--fields",
+        nargs="+",
+        default=[],
+        metavar="FIELD",
+        help="The metadata fields to be returned.",
+    )
+
+
+def run_data(args: argparse.Namespace) -> None:
+    data_subcommand = args.data_subcommand
+    del args.data_subcommand
+    if data_subcommand == "search":
+        data_search(**vars(args))

wigwam/data_commands.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Dict, Iterable, List
+
+from .search import filtered_file_search, names_only_search, search_file
+
+
+def data_search(
+    data_file: Path,
+    tags: Iterable[Iterable[str]],
+    names: Iterable[str],
+    fields: Iterable[str],
+    all: bool = False,
+    print_output: bool = True,
+) -> List[Dict[str, str | Dict[str, str]]]:
+    """
+    Query a file database for items.
+
+    Parameters
+    ----------
+    data_file : Path
+        The name of the database file.
+    tags : Iterable[Iterable[str]]
+        A set of sets of tags - this function will return the union of items that have
+        all of any of the sets of tags passed in.
+    names : Iterable[str]
+        A list of names of data items to return.
+    fields : Iterable[str]
+        The set of fields to be returned on the data items. This should be a strict
+        subset of the fields present on the items. Fields not included in this parameter
+        will be filtered from the items prior to returning them.
+    all : bool, optional
+        If true, return all of the items in the database. Defaults to False
+
+    Returns
+    -------
+    List[Dict[str, str | Dict[str, str]]]
+        The items returned by the query, in dictionary format.
+    """
+    if all:
+        if (any(True for _ in names)) or (any(True for _ in tags)):
+            print("'all' cannot be used in conjunction with 'tags' or 'names'.")
+            exit()
+
+    if fields == []:
+        return search_file(tags=tags, names=names, filename=data_file, all=all)
+
+    filtered_search = filtered_file_search(
+        fields=fields, names=names, tags=tags, filename=data_file, all=all
+    )
+
+    if print_output:
+        print(json.dumps(filtered_search, indent=2))
+
+    return filtered_search
+
+
+def data_names(
+    data_file: Path,
+    tags: Iterable[Iterable[str]],
+    names: Iterable[str],
+    all: bool = False,
+) -> List[str]:
+    """
+    Query a database file and return the names of all data items that match the query.
+
+    Parameters
+    ----------
+    data_file : Path
+        The path to the database file.
+    tags : Iterable[Iterable[str]]
+        A set of sets of tags - this function will return the union of items that have
+        all of any of the sets of tags passed in.
+    names : Iterable[str]
+        A list of names of data items to return.
+    all : bool, optional
+        If true, return all of the items in the database. Defaults to False
+
+    Returns
+    -------
+    List[str]
+        A list of the names of items that were returned by the query.
+    """
+    if all:
+        if (any(True for _ in names)) or (any(True for _ in tags)):
+            print("'all' cannot be used in conjunction with 'tags' or 'names'.")
+            exit()
+
+    return names_only_search(tags=tags, names=names, filename=data_file, all=all)

wigwam/defaults.py

@@ -14,3 +14,8 @@ def install_prefix() -> Path:
 def build_prefix() -> Path:
     """Returns the build system's default build prefix path."""
     return Path("/tmp/build")
+
+
+def default_workflowdata_path() -> Path:
+    """The default workflowdata.json path"""
+    return Path("workflowdata.json")