WIP: refactor base endpoints

src/client/qcc/cli/static.py

@@ -13,7 +13,6 @@
 from .codes import QCCExitCodes
 from .utils import parse_input_json_lines, parse_json_parameter
 
-
 app = typer.Typer(name="static", no_args_is_help=True)
 
 

src/client/qcc/cli/status.py

@@ -5,10 +5,11 @@
 from typing import Annotated, List, Optional
 
 import typer
+from rich import print
 
 from ..client import QCC
 from .api import async_run_api_query
-
+from .utils import parse_json_parameter
 
 app = typer.Typer(name="status", no_args_is_help=True)
 
@@ -29,6 +30,7 @@ async def statuses():
 
     async_run_api_query(statuses)
 
+
 @app.command()
 def create(
     ctx: typer.Context,
@@ -47,3 +49,7 @@ def create(
     expects your JSON string on a single row.
     """
     client: QCC = ctx.obj
+    data = parse_json_parameter("status", status, interactive)
+    created = async_run_api_query(client.static.create, data)
+    print("[green]Created statique successfully.[/green]")
+    print(created)

src/client/qcc/cli/utils.py

@@ -1,7 +1,7 @@
 """QualiCharge API client CLI: utils."""
 
 import json
-from typing import Generator, List, TextIO
+from typing import Generator, TextIO
 
 import click
 import typer

src/client/qcc/client.py

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

src/client/qcc/endpoints/base.py

@@ -0,0 +1,84 @@
+"""QualiCharge API client endpoints base."""
+
+import logging
+from abc import ABC, abstractmethod
+from typing import AsyncIterator, Sequence, overload
+
+import httpx
+
+from qcc.conf import settings
+from qcc.exceptions import APIRequestError
+from qcc.http import HTTPClient
+
+logger = logging.getLogger(__name__)
+
+
+class BaseCreateEndpoint:
+    """Base create endpoint."""
+
+    endpoint: str
+
+    def __init__(self, client: HTTPClient) -> None:
+        """Set /auth endpoints HTTP client."""
+        self.client = client
+
+    async def create(self, obj: dict) -> dict:
+        """Query the /{endpoint}/ endpoint (POST)."""
+        response = await self.client.post(f"{self.endpoint}/", json=obj)
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as err:
+            raise APIRequestError(response.json()) from err
+
+        return response.json()
+
+    async def bulk(
+        self,
+        objs: Sequence[dict],
+        chunk_size: int = settings.API_BULK_CREATE_MAX_SIZE,
+        ignore_errors: bool = False,
+    ) -> int:
+        """Query the /{endpoint}/bulk endpoint (POST)."""
+        chunk: list = []
+        n_created = 0
+
+        async def send_chunk(client, chunk: list[dict]) -> int:
+            """Submit a chunk to the API."""
+            response = await client.post(f"{self.endpoint}/bulk", json=chunk)
+            try:
+                response.raise_for_status()
+            except httpx.HTTPStatusError as err:
+                if ignore_errors:
+                    logger.debug("Ignored chunk: %s", chunk)
+                    logger.warning("Ignored query error: %s", response)
+                    return 0
+                raise APIRequestError(response.json()) from err
+            return response.json()["size"]
+
+        for obj in objs:
+            chunk.append(obj)
+            if len(chunk) == chunk_size:
+                n_created += await send_chunk(self.client, chunk)
+                chunk = []
+
+        if len(chunk):
+            n_created += await send_chunk(self.client, chunk)
+        return n_created
+
+
+class BaseEndpoint(ABC, BaseCreateEndpoint):
+    """Base endpoint CRUD actions."""
+
+    @abstractmethod
+    def list(self) -> AsyncIterator[dict]:
+        """Query the /{endpoint}/ endpoint (GET)."""
+
+    async def read(self, id_: str) -> dict:
+        """Query the /{endpoint}/{id_} endpoint (GET)."""
+        response = await self.client.get(f"{self.endpoint}/{id_}")
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as err:
+            raise APIRequestError(response.json()) from err
+
+        return response.json()

src/client/qcc/endpoints/dynamic.py

@@ -6,18 +6,18 @@
 
 import httpx
 
-from qcc.exceptions import APIRequestError
-from qcc.http import HTTPClient
+from qcc.endpoints.base import BaseCreateEndpoint
+
+from ..exceptions import APIRequestError
+from .base import BaseEndpoint
 
 logger = logging.getLogger(__name__)
 
 
-class Status:
+class Status(BaseEndpoint):
     """/dynamique/status endpoints."""
 
-    def __init__(self, client: HTTPClient) -> None:
-        """Set /auth endpoints HTTP client."""
-        self.client = client
+    endpoint: str = "/dynamique/status"
 
     async def list(
         self,
@@ -36,11 +36,28 @@ async def list(
             if p[1] is not None
         )
 
-        response = await self.client.get("/dynamique/status/", params=params)
+        response = await self.client.get(f"{self.endpoint}/", params=params)
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as err:
+            raise APIRequestError(response.json()) from err
+
+        for status in response.json():
+            yield status
+
+    async def history(self, id_: str) -> AsyncIterator[dict]:
+        """Query the /{endpoint}/{id_}/history endpoint (GET)."""
+        response = await self.client.get(f"{self.endpoint}/{id_}/history")
         try:
             response.raise_for_status()
         except httpx.HTTPStatusError as err:
             raise APIRequestError(response.json()) from err
 
         for status in response.json():
             yield status
+
+
+class Session(BaseCreateEndpoint):
+    """/dynamique/session endpoints."""
+
+    endpoint: str = "/dynamique/session"

src/client/qcc/endpoints/static.py

@@ -1,28 +1,25 @@
 """QualiCharge API client static endpoints."""
 
 import logging
-from typing import AsyncIterator, Sequence
+from typing import AsyncIterator
 
 import httpx
 
-from qcc.conf import settings
-from qcc.exceptions import APIRequestError
-from qcc.http import HTTPClient
+from ..exceptions import APIRequestError
+from .base import BaseEndpoint
 
 logger = logging.getLogger(__name__)
 
 
-class Static:
+class Static(BaseEndpoint):
     """/statique endpoints."""
 
-    def __init__(self, client: HTTPClient) -> None:
-        """Set /auth endpoints HTTP client."""
-        self.client = client
+    endpoint: str = "/statique"
 
     async def list(self) -> AsyncIterator[dict]:
         """Query the /statique/ endpoint (GET)."""
 
-        async def get_statiques(url="/statique/"):
+        async def get_statiques(url=f"{self.endpoint}/"):
             """Get statique items."""
             response = await self.client.get(url)
             try:
@@ -39,67 +36,12 @@ async def get_statiques(url="/statique/"):
         async for statique in get_statiques():
             yield statique
 
-    async def create(self, statique: dict) -> dict:
-        """Query the /statique/ endpoint (POST)."""
-        response = await self.client.post("/statique/", json=statique)
+    async def update(self, id_: str, obj: dict) -> dict:
+        """Query the /{endpoint}/{id_} endpoint (PUT)."""
+        response = await self.client.put(f"{self.endpoint}/{id_}", json=obj)
         try:
             response.raise_for_status()
         except httpx.HTTPStatusError as err:
             raise APIRequestError(response.json()) from err
 
         return response.json()
-
-    async def read(self, id_pdc_itinerance: str) -> dict:
-        """Query the /statique/{id_pdc_itinerance} endpoint (GET)."""
-        response = await self.client.get(f"/statique/{id_pdc_itinerance}")
-        try:
-            response.raise_for_status()
-        except httpx.HTTPStatusError as err:
-            raise APIRequestError(response.json()) from err
-
-        return response.json()
-
-    async def update(self, id_pdc_itinerance: str, statique: dict) -> dict:
-        """Query the /statique/{id_pdc_itinerance} endpoint (PUT)."""
-        response = await self.client.put(
-            f"/statique/{id_pdc_itinerance}", json=statique
-        )
-        try:
-            response.raise_for_status()
-        except httpx.HTTPStatusError as err:
-            raise APIRequestError(response.json()) from err
-
-        return response.json()
-
-    async def bulk(
-        self,
-        statiques: Sequence[dict],
-        chunk_size: int = settings.API_BULK_CREATE_MAX_SIZE,
-        ignore_errors: bool = False,
-    ) -> int:
-        """Query the /statique/bulk endpoint (POST)."""
-        chunk: list = []
-        n_created = 0
-
-        async def send_chunk(client, chunk: list[dict]) -> int:
-            """Submit a chunk to the API."""
-            response = await client.post("/statique/bulk", json=chunk)
-            try:
-                response.raise_for_status()
-            except httpx.HTTPStatusError as err:
-                if ignore_errors:
-                    logger.debug("Ignored chunk: %s", chunk)
-                    logger.warning("Ignored query error: %s", response)
-                    return 0
-                raise APIRequestError(response.json()) from err
-            return response.json()["size"]
-
-        for statique in statiques:
-            chunk.append(statique)
-            if len(chunk) == chunk_size:
-                n_created += await send_chunk(self.client, chunk)
-                chunk = []
-
-        if len(chunk):
-            n_created += await send_chunk(self.client, chunk)
-        return n_created

src/client/tests/endpoints/test_dynamic.py

@@ -1,6 +1,7 @@
 """Tests for the qcc.endpoints.dynamic module."""
 
 from datetime import datetime
+
 import pytest
 
 from qcc.endpoints.dynamic import Status