235 lines
9.4 KiB
Python
235 lines
9.4 KiB
Python
import argparse
|
|
import json
|
|
import sys
|
|
|
|
from .errors import TrackerError # type: ignore[reportMissingImports]
|
|
from .models import Envelope, RetryPolicy # type: ignore[reportMissingImports]
|
|
from .service import Tracker # type: ignore[reportMissingImports]
|
|
|
|
|
|
class JsonArgumentParser(argparse.ArgumentParser):
|
|
def error(self, message):
|
|
raise TrackerError("invalid_input", message)
|
|
|
|
|
|
def _parse_attempts(value):
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError) as error:
|
|
raise TrackerError("invalid_input", f"invalid retry attempt count: {value}") from error
|
|
|
|
|
|
def _take_global_options(argv):
|
|
remaining = []
|
|
provider = repo = None
|
|
attempts = 3
|
|
index = 0
|
|
while index < len(argv):
|
|
item = argv[index]
|
|
if item == "--provider" and index + 1 < len(argv):
|
|
provider = argv[index + 1]
|
|
index += 2
|
|
elif item.startswith("--provider="):
|
|
provider = item.split("=", 1)[1]
|
|
index += 1
|
|
elif item == "--repo" and index + 1 < len(argv):
|
|
repo = argv[index + 1]
|
|
index += 2
|
|
elif item.startswith("--repo="):
|
|
repo = item.split("=", 1)[1]
|
|
index += 1
|
|
elif item == "--retry-attempts" and index + 1 < len(argv):
|
|
attempts = _parse_attempts(argv[index + 1])
|
|
index += 2
|
|
elif item.startswith("--retry-attempts="):
|
|
attempts = _parse_attempts(item.split("=", 1)[1])
|
|
index += 1
|
|
else:
|
|
remaining.append(item)
|
|
index += 1
|
|
return remaining, provider, repo, attempts
|
|
|
|
|
|
def _add_resource_commands(subparsers, kind):
|
|
resource = subparsers.add_parser(kind)
|
|
commands = resource.add_subparsers(dest="action", required=True)
|
|
|
|
create = commands.add_parser("create")
|
|
create.add_argument("--title", required=True)
|
|
create.add_argument("--body", default="")
|
|
create.add_argument("--label", action="append", default=[])
|
|
create.add_argument("--assignee", action="append", default=[])
|
|
if kind == "pr":
|
|
create.add_argument("--head")
|
|
create.add_argument("--base")
|
|
|
|
get = commands.add_parser("get")
|
|
get.add_argument("number", type=int)
|
|
get.add_argument("--no-comments", action="store_true")
|
|
if kind == "pr":
|
|
get.add_argument("--diff", action="store_true")
|
|
|
|
listing = commands.add_parser("list")
|
|
listing.add_argument("--state", default="open")
|
|
listing.add_argument("--label", action="append", default=[])
|
|
listing.add_argument("--limit", type=int, default=100)
|
|
if kind == "pr":
|
|
listing.add_argument("--external-only", action="store_true")
|
|
|
|
comment = commands.add_parser("comment")
|
|
comment.add_argument("number", type=int)
|
|
comment.add_argument("--body", required=True)
|
|
|
|
edit = commands.add_parser("edit")
|
|
edit.add_argument("number", type=int)
|
|
edit.add_argument("--title")
|
|
edit.add_argument("--body")
|
|
|
|
assign = commands.add_parser("assign")
|
|
assign.add_argument("number", type=int)
|
|
assign.add_argument("--user", required=True)
|
|
|
|
close = commands.add_parser("close")
|
|
close.add_argument("number", type=int)
|
|
close.add_argument("--explanation")
|
|
|
|
if kind == "pr":
|
|
diff = commands.add_parser("diff")
|
|
diff.add_argument("number", type=int)
|
|
|
|
return resource
|
|
|
|
|
|
def build_parser():
|
|
parser = JsonArgumentParser(prog="tracker")
|
|
commands = parser.add_subparsers(dest="resource", required=True)
|
|
_add_resource_commands(commands, "issue")
|
|
_add_resource_commands(commands, "pr")
|
|
|
|
labels = commands.add_parser("label")
|
|
label_commands = labels.add_subparsers(dest="action", required=True)
|
|
ensure = label_commands.add_parser("ensure")
|
|
ensure.add_argument("name")
|
|
ensure.add_argument("--color", default="ededed")
|
|
ensure.add_argument("--description")
|
|
add = label_commands.add_parser("add")
|
|
add.add_argument("kind", choices=("issue", "pr"))
|
|
add.add_argument("number", type=int)
|
|
add.add_argument("name")
|
|
remove = label_commands.add_parser("remove")
|
|
remove.add_argument("kind", choices=("issue", "pr"))
|
|
remove.add_argument("number", type=int)
|
|
remove.add_argument("name")
|
|
|
|
map_parser = commands.add_parser("map")
|
|
map_commands = map_parser.add_subparsers(dest="action", required=True)
|
|
map_create = map_commands.add_parser("create")
|
|
map_create.add_argument("--title", required=True)
|
|
map_create.add_argument("--body", default="")
|
|
map_create.add_argument("--label", action="append", default=[])
|
|
|
|
child = commands.add_parser("child")
|
|
child_commands = child.add_subparsers(dest="action", required=True)
|
|
child_create = child_commands.add_parser("create")
|
|
child_create.add_argument("map_number", type=int)
|
|
child_create.add_argument("--title", required=True)
|
|
child_create.add_argument("--type", dest="wayfinder_type", choices=("research", "prototype", "grilling", "task"), default="task")
|
|
child_create.add_argument("--body", default="")
|
|
child_create.add_argument("--label", action="append", default=[])
|
|
|
|
dependency = commands.add_parser("dependency")
|
|
dependency_commands = dependency.add_subparsers(dest="action", required=True)
|
|
dependency_add = dependency_commands.add_parser("add")
|
|
dependency_add.add_argument("child", type=int)
|
|
dependency_add.add_argument("blocker", type=int)
|
|
|
|
frontier = commands.add_parser("frontier")
|
|
frontier.add_argument("map_number", type=int)
|
|
|
|
claim = commands.add_parser("claim")
|
|
claim.add_argument("kind", choices=("issue", "pr"))
|
|
claim.add_argument("number", type=int)
|
|
claim.add_argument("--user")
|
|
|
|
resolve = commands.add_parser("resolve")
|
|
resolve.add_argument("kind", choices=("issue", "pr"))
|
|
resolve.add_argument("number", type=int)
|
|
resolve.add_argument("--answer", required=True)
|
|
resolve.add_argument("--map", dest="map_number", type=int)
|
|
|
|
reference = commands.add_parser("resolve-reference")
|
|
reference.add_argument("number", type=int)
|
|
return parser
|
|
|
|
|
|
def _dispatch(tracker, args):
|
|
resource = args.resource
|
|
if resource in ("issue", "pr"):
|
|
if args.action == "create":
|
|
method = tracker.create_issue if resource == "issue" else tracker.create_pr
|
|
kwargs = {"body": args.body, "labels": args.label, "assignees": args.assignee}
|
|
if resource == "pr":
|
|
kwargs.update(head=args.head, base=args.base)
|
|
return method(args.title, **kwargs)
|
|
if args.action == "get":
|
|
if resource == "issue":
|
|
return tracker.get_issue(args.number, comments=not args.no_comments)
|
|
return tracker.get_pr(args.number, comments=not args.no_comments, diff=args.diff)
|
|
if args.action == "list":
|
|
if resource == "issue":
|
|
return tracker.list_issues(state=args.state, labels=args.label, limit=args.limit)
|
|
return tracker.list_prs(state=args.state, limit=args.limit, external_only=args.external_only)
|
|
if args.action == "comment":
|
|
return tracker.comment(resource, args.number, args.body)
|
|
if args.action == "edit":
|
|
return (tracker.edit_issue if resource == "issue" else tracker.edit_pr)(args.number, title=args.title, body=args.body)
|
|
if args.action == "assign":
|
|
return tracker.assign(resource, args.number, args.user)
|
|
if args.action == "close":
|
|
return tracker.close(resource, args.number, explanation=args.explanation)
|
|
if args.action == "diff":
|
|
return tracker.diff(args.number)
|
|
if resource == "label":
|
|
if args.action == "ensure":
|
|
return tracker.ensure_label(args.name, color=args.color, description=args.description)
|
|
if args.action == "add":
|
|
return tracker.add_label(args.kind, args.number, args.name)
|
|
return tracker.remove_label(args.kind, args.number, args.name)
|
|
if resource == "map":
|
|
return tracker.create_map(args.title, body=args.body, labels=args.label)
|
|
if resource == "child":
|
|
return tracker.create_child(args.map_number, args.title, wayfinder_type=args.wayfinder_type, body=args.body, labels=args.label)
|
|
if resource == "dependency":
|
|
return tracker.add_dependency(args.child, args.blocker)
|
|
if resource == "frontier":
|
|
return tracker.frontier(args.map_number)
|
|
if resource == "claim":
|
|
return tracker.claim(args.kind, args.number, user=args.user)
|
|
if resource == "resolve":
|
|
return tracker.resolve(args.kind, args.number, args.answer, map_number=args.map_number)
|
|
return tracker.resolve_reference(args.number)
|
|
|
|
|
|
def main(argv=None):
|
|
argv = list(sys.argv[1:] if argv is None else argv)
|
|
try:
|
|
command_argv, provider, repo, attempts = _take_global_options(argv)
|
|
args = build_parser().parse_args(command_argv)
|
|
tracker = Tracker(provider=provider, repo=repo, retry=RetryPolicy(attempts=attempts))
|
|
envelope = _dispatch(tracker, args)
|
|
except TrackerError as error:
|
|
operation = error.operation
|
|
if operation is None:
|
|
operation = "cli"
|
|
error.operation = operation
|
|
envelope = Envelope(False, getattr(error, "provider", None), operation, error=error.to_dict()).to_dict()
|
|
except (ValueError, TypeError, OSError) as error:
|
|
failure = TrackerError("invalid_input", str(error), operation="cli")
|
|
envelope = Envelope(False, None, "cli", error=failure.to_dict()).to_dict()
|
|
print(json.dumps(envelope, sort_keys=True))
|
|
return 0 if envelope.get("ok") else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|