Merge branch 'develop' into kl/hp-tests

cvat-cli/src/cvat_cli/__main__.py

@@ -1,79 +1,37 @@
 # Copyright (C) 2020-2022 Intel Corporation
-# Copyright (C) 2022 CVAT.ai Corporation
+# Copyright (C) 2022-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
+import argparse
 import logging
 import sys
-from http.client import HTTPConnection
-from types import SimpleNamespace
-from typing import List
 
 import urllib3.exceptions
 from cvat_sdk import exceptions
-from cvat_sdk.core.client import Client, Config
 
-from cvat_cli.cli import CLI
-from cvat_cli.parser import get_action_args, make_cmdline_parser
+from ._internal.commands import COMMANDS
+from ._internal.common import build_client, configure_common_arguments, configure_logger
+from ._internal.utils import popattr
 
 logger = logging.getLogger(__name__)
 
 
-def configure_logger(level):
-    formatter = logging.Formatter(
-        "[%(asctime)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", style="%"
-    )
-    handler = logging.StreamHandler(sys.stdout)
-    handler.setFormatter(formatter)
-    logger.addHandler(handler)
-    logger.setLevel(level)
-    if level <= logging.DEBUG:
-        HTTPConnection.debuglevel = 1
+def main(args: list[str] = None):
+    parser = argparse.ArgumentParser(description=COMMANDS.description)
+    configure_common_arguments(parser)
+    COMMANDS.configure_parser(parser)
 
-
-def build_client(parsed_args: SimpleNamespace, logger: logging.Logger) -> Client:
-    config = Config(verify_ssl=not parsed_args.insecure)
-
-    url = parsed_args.server_host
-    if parsed_args.server_port:
-        url += f":{parsed_args.server_port}"
-
-    client = Client(
-        url=url,
-        logger=logger,
-        config=config,
-        check_server_version=False,  # version is checked after auth to support versions < 2.3
-    )
-
-    client.organization_slug = parsed_args.organization
-
-    return client
-
-
-def main(args: List[str] = None):
-    actions = {
-        "create": CLI.tasks_create,
-        "delete": CLI.tasks_delete,
-        "ls": CLI.tasks_list,
-        "frames": CLI.tasks_frames,
-        "dump": CLI.tasks_dump,
-        "upload": CLI.tasks_upload,
-        "export": CLI.tasks_export,
-        "import": CLI.tasks_import,
-        "auto-annotate": CLI.tasks_auto_annotate,
-    }
-    parser = make_cmdline_parser()
     parsed_args = parser.parse_args(args)
-    configure_logger(parsed_args.loglevel)
 
-    with build_client(parsed_args, logger=logger) as client:
-        action_args = get_action_args(parser, parsed_args)
-        try:
-            cli = CLI(client=client, credentials=parsed_args.auth)
-            actions[parsed_args.action](cli, **vars(action_args))
-        except (exceptions.ApiException, urllib3.exceptions.HTTPError) as e:
-            logger.critical(e)
-            return 1
+    configure_logger(logger, parsed_args)
+
+    try:
+        with build_client(parsed_args, logger=logger) as client:
+            popattr(parsed_args, "_executor")(client, **vars(parsed_args))
+    except (exceptions.ApiException, urllib3.exceptions.HTTPError) as e:
+        logger.critical(e)
+        return 1
 
     return 0
 

cvat-cli/src/cvat_cli/_internal/command_base.py

@@ -0,0 +1,53 @@
+# Copyright (C) 2024 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+import argparse
+import types
+from collections.abc import Mapping
+from typing import Callable, Protocol
+
+
+class Command(Protocol):
+    @property
+    def description(self) -> str: ...
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None: ...
+
+    # The exact parameters accepted by `execute` vary between commands,
+    # so we're forced to declare it like this instead of as a method.
+    @property
+    def execute(self) -> Callable[..., None]: ...
+
+
+class CommandGroup:
+    def __init__(self, *, description: str) -> None:
+        self._commands: dict[str, Command] = {}
+        self.description = description
+
+    def command_class(self, name: str):
+        def decorator(cls: type):
+            self._commands[name] = cls()
+            return cls
+
+        return decorator
+
+    def add_command(self, name: str, command: Command) -> None:
+        self._commands[name] = command
+
+    @property
+    def commands(self) -> Mapping[str, Command]:
+        return types.MappingProxyType(self._commands)
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        subparsers = parser.add_subparsers(required=True)
+
+        for name, command in self._commands.items():
+            subparser = subparsers.add_parser(name, description=command.description)
+            subparser.set_defaults(_executor=command.execute)
+            command.configure_parser(subparser)
+
+    def execute(self) -> None:
+        # It should be impossible for a command group to be executed,
+        # because configure_parser requires that a subcommand is specified.
+        assert False, "unreachable code"