WIP: refactor stdin CLI utilities

src/client/qcc/cli/static.py

@@ -1,7 +1,7 @@
 """QualiCharge API client CLI: static."""
 
 import json
-from typing import Generator, Optional
+from typing import Optional
 
 import click
 import typer
@@ -11,6 +11,8 @@
 from ..client import QCC
 from .api import async_run_api_query
 from .codes import QCCExitCodes
+from .utils import parse_input_json_lines, parse_json_parameter
+
 
 app = typer.Typer(name="static", no_args_is_help=True)
 
@@ -45,23 +47,8 @@ def create(
     expects your JSON string on a single row.
     """
     client: QCC = ctx.obj
-    if not statique and interactive:
-        statique = click.get_text_stream("stdin").readline()
-
-    if statique is None:
-        print(
-            "[red]A statique object is required either from stdin or as an option[/red]"
-        )
-        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION)
-
-    try:
-        data = json.loads(statique)
-    except json.JSONDecodeError as err:
-        print("[red]Invalid JSON input string[/red]")
-        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION) from err
-
+    data = parse_json_parameter("statique", statique, interactive)
     created = async_run_api_query(client.static.create, data)
-
     print("[green]Created statique successfully.[/green]")
     print(created)
 
@@ -93,20 +80,7 @@ def update(
     expects your JSON string on a single row.
     """
     client: QCC = ctx.obj
-    if not statique and interactive:
-        statique = click.get_text_stream("stdin").readline()
-
-    if statique is None:
-        print(
-            "[red]A statique object is required either from stdin or as an option[/red]"
-        )
-        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION)
-
-    try:
-        data = json.loads(statique)
-    except json.JSONDecodeError as err:
-        print("[red]Invalid JSON input string[/red]")
-        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION) from err
+    data = parse_json_parameter("statique", statique, interactive)
 
     if "id_pdc_itinerance" not in data:
         print("[red]Statique object requires an `id_pdc_itinerance` field[/red]")
@@ -130,22 +104,9 @@ def bulk(
     """
     client: QCC = ctx.obj
 
-    def parse_input_json_lines(lines) -> Generator[dict, None, None]:
-        """Read and JSON parse stdin line by line."""
-        for statique in lines:
-            try:
-                data = json.loads(statique)
-            except json.JSONDecodeError as err:
-                if ignore_errors:
-                    print(f"[orange]Ignored invalid line:[/orange]\n{statique}")
-                    continue
-                print("[red]Invalid JSON input string[/red]")
-                raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION) from err
-            yield data
-
     n_created = async_run_api_query(
         client.static.bulk,
-        parse_input_json_lines(click.get_text_stream("stdin")),
+        parse_input_json_lines(click.get_text_stream("stdin"), ignore_errors),
         chunk_size,
         ignore_errors,
     )

src/client/qcc/cli/utils.py

@@ -0,0 +1,50 @@
+"""QualiCharge API client CLI: utils."""
+
+import json
+from typing import Generator, List, TextIO
+
+import click
+import typer
+from rich import print
+
+from .codes import QCCExitCodes
+
+
+def parse_json_parameter(name: str, parameter: str, interactive: bool) -> dict:
+    """Read and JSON parse parameter from option or stdin."""
+    # Get parameter value from stdin if empty
+    if not parameter and interactive:
+        parameter = click.get_text_stream("stdin").readline()
+
+    if parameter is None:
+        print(
+            (
+                f"[red]A {name} object is required "
+                "either from stdin or as an option[/red]"
+            )
+        )
+        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION)
+
+    # Parse parameter as JSON
+    try:
+        data = json.loads(parameter)
+    except json.JSONDecodeError as err:
+        print("[red]Invalid JSON input string[/red]")
+        raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION) from err
+    return data
+
+
+def parse_input_json_lines(
+    lines: TextIO, ignore_errors: bool
+) -> Generator[dict, None, None]:
+    """Read and JSON parse stdin line by line."""
+    for line in lines:
+        try:
+            data = json.loads(line)
+        except json.JSONDecodeError as err:
+            if ignore_errors:
+                print(f"[orange]Ignored invalid line:[/orange]\n{line}")
+                continue
+            print("[red]Invalid JSON input string[/red]")
+            raise typer.Exit(QCCExitCodes.PARAMETER_EXCEPTION) from err
+        yield data