Merge pull request #72 from rossumai/refactor/extract-arguments

Refactor/extract arguments + remove the necessity to specify file type of loaded schema

elisctl/argument.py

@@ -1,15 +1,25 @@
-import json
-
 import functools
+import json
+from typing import IO, Optional, Callable, Iterable
 
 import click
-from typing import IO
 
-id_ = click.argument("id_", metavar="ID", type=int)
+from elisctl.common import schema_content_factory
+from elisctl.lib import split_dict_params
+
 name = click.argument("name", type=str)
 
 
-def schema_file(command):
+def id_(command: Optional[Callable] = None, **kwargs):
+    default_kwargs = {"type": int, "metavar": "ID"}
+    kwargs = {**default_kwargs, **kwargs}
+    decorator = click.argument("id_", **kwargs)
+    if command is None:
+        return decorator
+    return decorator(command)
+
+
+def schema_file(command: Callable):
     click.argument("schema_file", type=click.File("rb"))(command)
 
     @functools.wraps(command)
@@ -18,3 +28,29 @@ def wrapped(ctx: click.Context, schema_file: IO[str], *args, **kwargs):
         return command(ctx, *args, **kwargs)
 
     return wrapped
+
+
+def datapoint_parameters(command: Callable):
+    click.argument("datapoint_parameters", nargs=-1, type=str)(command)
+
+    @functools.wraps(command)
+    def wrapped(*args, datapoint_parameters: Iterable[str], **kwargs):
+        try:
+            split_params = split_dict_params(datapoint_parameters)
+        except ValueError as e:
+            raise click.BadArgumentUsage("Expecting <key>=<value> pairs.") from e
+        return command(*args, datapoint_parameters=dict(split_params), **kwargs)
+
+    return wrapped
+
+
+def schema_content_file(command: Optional[Callable] = None, **kwargs):
+    default_kwargs = {"type": click.File("rb"), "metavar": "FILE"}
+    kwargs = {**default_kwargs, **kwargs}
+    decorator = click.argument("schema_content_file_", **kwargs)
+    if command is None:
+        return decorator
+    return decorator(command)
+
+
+schema_content = schema_content_factory(schema_content_file)

elisctl/common.py

@@ -0,0 +1,45 @@
+import json
+import functools
+from typing import Callable, IO, Optional
+
+import click
+
+from elisctl.schema.xlsx import SchemaContent, XlsxToSchema
+
+
+def schema_content_factory(file_decorator):
+    def schema_content(command: Optional[Callable] = None, **file_decorator_kwargs):
+        def _load_func(schema_content_file_: Optional[IO[bytes]]) -> Optional[SchemaContent]:
+            if schema_content_file_ is None:
+                return None
+
+            errors = []
+            for load_func in (json.load, XlsxToSchema().convert):
+                try:
+                    return load_func(schema_content_file_)  # type: ignore
+                except Exception as e:
+                    errors.append(str(e))
+                    schema_content_file_.seek(0)
+
+            raise NotImplementedError(", ".join(errors))
+
+        def decorator(command_: Callable):
+            @file_decorator(**file_decorator_kwargs)
+            @functools.wraps(command_)
+            def wrapped(*args, schema_content_file_: Optional[IO[bytes]], **kwargs):
+                try:
+                    schema_content_ = _load_func(schema_content_file_)
+                except Exception as e:
+                    raise click.ClickException(
+                        f"File {schema_content_file_} could not be loaded. Because of {e}"
+                    ) from e
+
+                return command_(*args, schema_content=schema_content_, **kwargs)
+
+            return wrapped
+
+        if command is None:
+            return decorator
+        return decorator(command)
+
+    return schema_content

elisctl/lib/__init__.py

@@ -4,12 +4,19 @@
 from contextlib import suppress
 from dataclasses import dataclass
 
-from typing import Iterable, Iterator, Tuple, Union
+from typing import Iterable, Iterator, Tuple, Union, Dict, List
+
+# todo: use TypedDict if available https://www.python.org/dev/peps/pep-0589/
+# todo: use Recursion (see the following) when https://github.com/python/mypy/issues/731 is ready
+# DataPointDictItem = Union[str, int, "DataPointDict", None, "DataPoints"]
+DataPointDictItem = Union[str, int, dict, None, list]
+DataPointDict = Dict[str, DataPointDictItem]
+DataPoints = List[DataPointDict]
 
 
 def split_dict_params(
     datapoint_parameters: Iterable[str]
-) -> Iterator[Tuple[str, Union[str, int, dict, None, list]]]:
+) -> Iterator[Tuple[str, DataPointDictItem]]:
     for param in datapoint_parameters:
         key, value = param.split("=", 1)
         with suppress(ValueError):

elisctl/main.py

@@ -1,9 +1,9 @@
 import click
 from click_shell import shell
 
+import elisctl.schema.commands
 from elisctl import (
     tools,
-    schema,
     csv,
     configure,
     user,
@@ -38,7 +38,7 @@ def entry_point(ctx: click.Context, profile: str) -> None:
 
 entry_point.add_command(tools.cli)
 entry_point.add_command(csv.cli)
-entry_point.add_command(schema.cli)
+entry_point.add_command(elisctl.schema.commands.cli)
 entry_point.add_command(user.cli)
 entry_point.add_command(workspace.cli)
 entry_point.add_command(queue.cli)

elisctl/option.py

@@ -2,6 +2,9 @@
 
 import click
 
+from elisctl.common import schema_content_factory
+
+
 organization = click.option(
     "-o", "--organization-id", type=int, help="Organization ID.", hidden=True
 )
@@ -21,14 +24,17 @@
 
 
 def schema_content_file(command: Optional[Callable] = None, **kwargs):
-    default_kwargs = {"type": click.File("rb"), "help": "Schema JSON file."}
+    default_kwargs = {"type": click.File("rb"), "help": "Schema file."}
     kwargs = {**default_kwargs, **kwargs}
-    decorator = click.option("-s", "--schema-content-file", **kwargs)
+    decorator = click.option("-s", "--schema-content-file", "schema_content_file_", **kwargs)
     if command is None:
         return decorator
     return decorator(command)
 
 
+schema_content = schema_content_factory(schema_content_file)
+
+
 def workspace_id(command: Optional[Callable] = None, **kwargs):
     default_kwargs = {"type": int, "help": "Workspace ID."}
     kwargs = {**default_kwargs, **kwargs}

elisctl/queue.py

@@ -1,5 +1,4 @@
-import json
-from typing import Optional, IO, Dict, Any
+from typing import Optional, Dict, Any, List
 
 import click
 from tabulate import tabulate
@@ -22,7 +21,7 @@ def cli() -> None:
 
 @cli.command(name="create", help="Create queue.")
 @click.argument("name")
-@option.schema_content_file(required=True)
+@option.schema_content(required=True)
 @option.email_prefix
 @option.bounce_email
 @option.workspace_id
@@ -32,14 +31,13 @@ def cli() -> None:
 def create_command(
     ctx: click.Context,
     name: str,
-    schema_content_file: IO[bytes],
+    schema_content: List[dict],
     email_prefix: Optional[str],
     bounce_email: Optional[str],
     workspace_id: Optional[int],
     connector_id: Optional[int],
     locale: Optional[str],
 ) -> None:
-    schema_content = json.load(schema_content_file)
     if email_prefix is not None and bounce_email is None:
         raise click.ClickException("Inbox cannot be created without specified bounce email.")
 
@@ -105,19 +103,19 @@ def delete_command(ctx: click.Context, id_: int) -> None:
 @cli.command(name="change", help="Change a queue.")
 @argument.id_
 @option.name
-@option.schema_content_file
+@option.schema_content
 @option.connector_id
 @locale_option
 @click.pass_context
 def change_command(
     ctx: click.Context,
     id_: int,
     name: Optional[str],
-    schema_content_file: Optional[IO[bytes]],
+    schema_content: Optional[List[dict]],
     connector_id: Optional[int],
     locale: Optional[str],
 ) -> None:
-    if not any([name, schema_content_file, connector_id, locale]):
+    if not any([name, schema_content, connector_id, locale]):
         return
 
     data: Dict[str, Any] = {}
@@ -132,9 +130,8 @@ def change_command(
         if connector_id is not None:
             data["connector"] = get_json(elis.get(f"connectors/{connector_id}"))["url"]
 
-        if schema_content_file is not None:
+        if schema_content is not None:
             name = name or elis.get_queue(id_)["name"]
-            schema_content = json.load(schema_content_file)
             schema_dict = elis.create_schema(f"{name} schema", schema_content)
             data["schema"] = schema_dict["url"]
 

elisctl/schema/__init__.py

@@ -1,45 +0,0 @@
-import json
-from typing import IO, Optional
-
-import click
-
-from elisctl import option
-from elisctl.lib.api_client import APIClient, get_json
-from elisctl.schema.xlsx import SchemaToXlsx
-from . import transform, upload
-
-
-@click.group("schema")
-def cli() -> None:
-    pass
-
-
-cli.add_command(transform.cli)
-cli.add_command(upload.upload_command)
-
-
-@cli.command(name="get", help="Download schema from ELIS.")
-@click.pass_context
-@click.argument("id_", metavar="ID", type=str)
-@click.option("--indent", default=2, type=int)
-@click.option("--ensure-ascii", is_flag=True, type=bool)
-@click.option("--format", "format_", default="json", type=click.Choice(["json", "xlsx"]))
-@option.output_file
-def download_command(
-    ctx: click.Context,
-    id_: str,
-    indent: int,
-    ensure_ascii: bool,
-    format_: str,
-    output_file: Optional[IO[str]],
-):
-    with APIClient(context=ctx.obj) as api_client:
-        schema_dict = get_json(api_client.get(f"schemas/{id_}"))
-    if format_ == "json":
-        schema_file = json.dumps(
-            schema_dict["content"], indent=indent, ensure_ascii=ensure_ascii, sort_keys=True
-        ).encode("utf-8")
-    else:
-        schema_file = SchemaToXlsx().convert(schema_dict["content"])
-
-    click.echo(schema_file, file=output_file, nl=False)

elisctl/schema/commands.py

@@ -0,0 +1,47 @@
+import json
+
+from typing import Optional, IO
+
+import click
+
+from elisctl import option
+from elisctl.lib.api_client import APIClient, get_json
+from . import upload
+from .xlsx import SchemaToXlsx
+from .transform import commands as transform
+
+
+@click.group("schema")
+def cli() -> None:
+    pass
+
+
+cli.add_command(transform.cli)
+cli.add_command(upload.upload_command)
+
+
+@cli.command(name="get", help="Download schema from ELIS.")
+@click.pass_context
+@click.argument("id_", metavar="ID", type=str)
+@click.option("--indent", default=2, type=int)
+@click.option("--ensure-ascii", is_flag=True, type=bool)
+@click.option("--format", "format_", default="json", type=click.Choice(["json", "xlsx"]))
+@option.output_file
+def download_command(
+    ctx: click.Context,
+    id_: str,
+    indent: int,
+    ensure_ascii: bool,
+    format_: str,
+    output_file: Optional[IO[str]],
+):
+    with APIClient(context=ctx.obj) as api_client:
+        schema_dict = get_json(api_client.get(f"schemas/{id_}"))
+    if format_ == "json":
+        schema_file = json.dumps(
+            schema_dict["content"], indent=indent, ensure_ascii=ensure_ascii, sort_keys=True
+        ).encode("utf-8")
+    else:
+        schema_file = SchemaToXlsx().convert(schema_dict["content"])
+
+    click.echo(schema_file, file=output_file, nl=False)