Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Accept WDL file as first argument to support shebang. #164

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 35 additions & 10 deletions WDL/CLI.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,26 +17,36 @@
import WDL
import WDL.Lint
from . import values_from_json, values_to_json
import collections

quant_warning = False


def main(args=None):
command_dict = collections.OrderedDict(
check=fill_check_subparser, cromwell=fill_cromwell_subparser
)
parser = ArgumentParser()
parser.add_argument(
"--version",
nargs=0,
action=PipVersionAction,
help="show miniwdl package version information",
)
args = args if args is not None else sys.argv[1:]
# If the first arg is a file (shebang was used),
# add it as the first argparse argument and skip adding it in the subparsers
is_file_first = check_is_file_first(args, command_dict)
if is_file_first:
parser.add_argument("uri", help="This WDL file")
subparsers = parser.add_subparsers()
subparsers.required = True
subparsers.dest = "command"
fill_common(fill_check_subparser(subparsers))
fill_common(fill_cromwell_subparser(subparsers))
for command, fill_subparser in command_dict.items():
fill_common(fill_subparser(subparsers, skip_uri=is_file_first))

argcomplete.autocomplete(parser)
args = parser.parse_args(args if args is not None else sys.argv[1:])
args = parser.parse_args(args)

try:
if args.command == "check":
Expand All @@ -63,6 +73,15 @@ def main(args=None):
sys.exit(2)


def check_is_file_first(args, command_dict):
if len(args) < 1:
return False
first = args[0]
if first in command_dict:
return False
return os.path.exists(first)


class PipVersionAction(Action):
def __call__(self, parser, namespace, values, option_string=None):
try:
Expand Down Expand Up @@ -91,13 +110,14 @@ def fill_common(subparser):
subparser.add_argument("--debug", action="store_true", help="show full exception traceback")


def fill_check_subparser(subparsers):
def fill_check_subparser(subparsers, skip_uri=False):
check_parser = subparsers.add_parser(
"check", help="Load and typecheck a WDL document; show an outline with lint warnings"
)
check_parser.add_argument(
"uri", metavar="URI", type=str, nargs="+", help="WDL document filename/URI"
)
if not skip_uri:
check_parser.add_argument(
"uri", metavar="URI", type=str, nargs="+", help="WDL document filename/URI"
)
check_parser.add_argument(
"--no-shellcheck",
dest="shellcheck",
Expand All @@ -111,7 +131,9 @@ def check(uri=None, path=None, check_quant=True, shellcheck=True, **kwargs):
# Load the document (read, parse, and typecheck)
if not shellcheck:
WDL.Lint._shellcheck_available = False

# Accept also just one URI by wrapping it in a singleton list
if isinstance(uri, str):
uri = [uri]
for uri1 in uri or []:
doc = WDL.load(uri1, path or [], check_quant=check_quant, import_uri=import_uri)

Expand Down Expand Up @@ -251,11 +273,14 @@ def import_uri(uri):
return glob.glob(dn + "/*")[0]


def fill_cromwell_subparser(subparsers):
def fill_cromwell_subparser(subparsers, skip_uri=False):
cromwell_parser = subparsers.add_parser(
"cromwell", help="Run workflow locally using Cromwell " + CROMWELL_VERSION
)
cromwell_parser.add_argument("uri", metavar="URI", type=str, help="WDL document filename/URI")
if not skip_uri:
cromwell_parser.add_argument(
"uri", metavar="URI", type=str, help="WDL document filename/URI"
)
cromwell_parser.add_argument(
"inputs",
metavar="input_key=value",
Expand Down