✨(client) add dynamic endpoints and CLI commands

We are now able to query /dynamique API endpoints using the client but
also the CLI, yay!

Author: jmaupetit

src/client/qcc/cli/__init__.py

@@ -7,11 +7,13 @@
 
 from ..client import QCC
 from ..conf import settings
-from . import auth, static
+from . import auth, session, static, status
 
 app = typer.Typer(name="qcc", no_args_is_help=True)
 app.add_typer(auth.app)
 app.add_typer(static.app)
+app.add_typer(status.app)
+app.add_typer(session.app)
 
 
 @app.callback()

src/client/qcc/cli/api.py

@@ -0,0 +1,21 @@
+"""QualiCharge API client CLI: api."""
+
+from typing import Any
+
+import typer
+from anyio import run
+from rich import print
+
+from ..exceptions import APIRequestError
+from .codes import QCCExitCodes
+
+
+def async_run_api_query(*args) -> Any:
+    """An anyio.run wrapper to handle APIRequestError."""
+    try:
+        return_value = run(*args)
+    except APIRequestError as err:
+        print("[red]An error occurred while querying the API! More details follow.")
+        print(err.args[0])
+        raise typer.Exit(QCCExitCodes.API_EXCEPTION) from err
+    return return_value

src/client/qcc/cli/session.py

@@ -0,0 +1,59 @@
+"""QualiCharge API client CLI: statuc."""
+
+from typing import Annotated, Optional
+
+import click
+import typer
+from rich import print
+
+from ..client import QCC
+from .api import async_run_api_query
+from .utils import parse_input_json_lines, parse_json_parameter
+
+app = typer.Typer(name="session", no_args_is_help=True)
+
+
+@app.command()
+def create(
+    ctx: typer.Context,
+    session: Optional[str] = None,
+    interactive: Annotated[
+        bool, typer.Option(help="Read session from standard input (JSON string)")
+    ] = True,
+):
+    """Create a charging point session.
+
+    You can submit your session entry to create as a JSON string argument for
+    the `--status` option. Without `--status` option (but with `--interactive`)
+    the command will read and parse the standard input as a JSON string.
+
+    Note that when using the `--interactive` option (active by default), the command
+    expects your JSON string on a single row.
+    """
+    client: QCC = ctx.obj
+    data = parse_json_parameter("session", session, interactive)  # type: ignore[arg-type]
+    created = async_run_api_query(client.session.create, data)
+    print("[green]Created session successfully.[/green]")
+    print(created)
+
+
+@app.command()
+def bulk(
+    ctx: typer.Context,
+    chunk_size: int = 10,
+    ignore_errors: bool = False,
+):
+    """Bulk create new sessions.
+
+    Sessions will be read from the standard input (one JSON per line).
+    """
+    client: QCC = ctx.obj
+
+    n_created = async_run_api_query(
+        client.session.bulk,
+        parse_input_json_lines(click.get_text_stream("stdin"), ignore_errors),
+        chunk_size,
+        ignore_errors,
+    )
+
+    print(f"[green]Created {n_created} sessions successfully.[/green]")

src/client/qcc/cli/static.py

@@ -1,35 +1,24 @@
 """QualiCharge API client CLI: static."""
 
 import json
-from typing import Any, Generator, Optional
+from typing import Optional
 
 import click
 import typer
-from anyio import run
 from rich import print
 from typing_extensions import Annotated
 
 from ..client import QCC
-from ..exceptions import APIRequestError
+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)
 
 
-def async_run_api_query(*args) -> Any:
-    """An anyio.run wrapper to handle APIRequestError."""
-    try:
-        return_value = run(*args)
-    except APIRequestError as err:
-        print("[red]An error occurred while querying the API! More details follow.")
-        print(err.args[0])
-        raise typer.Exit(QCCExitCodes.API_EXCEPTION) from err
-    return return_value
-
-
 @app.command()
 def list(ctx: typer.Context):
-    """Get all static entries."""
+    """Get all statique entries."""
     client: QCC = ctx.obj
 
     async def statiques():
@@ -57,30 +46,15 @@ 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)  # type: ignore[arg-type]
     created = async_run_api_query(client.static.create, data)
-
     print("[green]Created statique successfully.[/green]")
     print(created)
 
 
 @app.command()
 def read(ctx: typer.Context, id_pdc_itinerance: str):
-    """Get all static entries."""
+    """Read a statique entry."""
     client: QCC = ctx.obj
 
     read = async_run_api_query(client.static.read, id_pdc_itinerance)
@@ -105,20 +79,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)  # type: ignore[arg-type]
 
     if "id_pdc_itinerance" not in data:
         print("[red]Statique object requires an `id_pdc_itinerance` field[/red]")
@@ -142,22 +103,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/status.py

@@ -0,0 +1,103 @@
+"""QualiCharge API client CLI: status."""
+
+import json
+from datetime import datetime
+from typing import Annotated, List, Optional
+
+import click
+import typer
+from rich import print
+
+from ..client import QCC
+from .api import async_run_api_query
+from .utils import parse_input_json_lines, parse_json_parameter
+
+app = typer.Typer(name="status", no_args_is_help=True)
+
+
+@app.command()
+def list(
+    ctx: typer.Context,
+    from_: Annotated[Optional[datetime], typer.Option("--from")] = None,
+    pdc: Optional[List[str]] = None,
+    station: Optional[List[str]] = None,
+):
+    """List charging points last known status."""
+    client: QCC = ctx.obj
+
+    async def statuses():
+        async for status in client.status.list(from_=from_, pdc=pdc, station=station):
+            typer.echo(json.dumps(status))
+
+    async_run_api_query(statuses)
+
+
+@app.command()
+def create(
+    ctx: typer.Context,
+    status: Optional[str] = None,
+    interactive: Annotated[
+        bool, typer.Option(help="Read status from standard input (JSON string)")
+    ] = True,
+):
+    """Create a charging point status.
+
+    You can submit your status entry to create as a JSON string argument for
+    the `--status` option. Without `--status` option (but with `--interactive`)
+    the command will read and parse the standard input as a JSON string.
+
+    Note that when using the `--interactive` option (active by default), the command
+    expects your JSON string on a single row.
+    """
+    client: QCC = ctx.obj
+    data = parse_json_parameter("status", status, interactive)  # type: ignore[arg-type]
+    created = async_run_api_query(client.status.create, data)
+    print("[green]Created status successfully.[/green]")
+    print(created)
+
+
+@app.command()
+def read(ctx: typer.Context, id_pdc_itinerance: str):
+    """Get charging point status."""
+    client: QCC = ctx.obj
+
+    read = async_run_api_query(client.status.read, id_pdc_itinerance)
+    typer.echo(json.dumps(read))
+
+
+@app.command()
+def history(
+    ctx: typer.Context,
+    id_pdc_itinerance: str,
+    from_: Annotated[Optional[datetime], typer.Option("--from")] = None,
+):
+    """Get charging point history."""
+    client: QCC = ctx.obj
+
+    async def statuses():
+        async for status in client.status.history(id_pdc_itinerance, from_=from_):
+            typer.echo(json.dumps(status))
+
+    async_run_api_query(statuses)
+
+
+@app.command()
+def bulk(
+    ctx: typer.Context,
+    chunk_size: int = 10,
+    ignore_errors: bool = False,
+):
+    """Bulk create new statuses.
+
+    Statuses will be read from the standard input (one JSON per line).
+    """
+    client: QCC = ctx.obj
+
+    n_created = async_run_api_query(
+        client.status.bulk,
+        parse_input_json_lines(click.get_text_stream("stdin"), ignore_errors),
+        chunk_size,
+        ignore_errors,
+    )
+
+    print(f"[green]Created {n_created} statuses successfully.[/green]")

src/client/qcc/cli/utils.py

@@ -0,0 +1,50 @@
+"""QualiCharge API client CLI: utils."""
+
+import json
+from typing import Generator, 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

src/client/qcc/client.py

@@ -1,6 +1,7 @@
 """QualiCharge API client."""
 
 from .endpoints.auth import Auth
+from .endpoints.dynamic import Session, Status
 from .endpoints.static import Static
 from .exceptions import ConfigurationError
 from .http import HTTPClient
@@ -26,3 +27,5 @@ def __init__(
         )
         self.auth = Auth(self.client)
         self.static = Static(self.client)
+        self.status = Status(self.client)
+        self.session = Session(self.client)