chore: Install Ariadne-codegen to generate a typed open hexa client (HEXA-1254) (#264)

Author: yolanfery

.github/workflows/ci.yml

@@ -17,6 +17,9 @@ jobs:
         with:
           python-version: 3.13
 
+      - name: Install Ariadne-codegen
+        run: pip install ariadne-codegen
+
       - uses: pre-commit/action@v3.0.1
   test:
     name: Run test suite

.pre-commit-config.yaml

@@ -4,4 +4,15 @@ repos:
   hooks:
     - id: ruff
       args: [ --fix, --exit-non-zero-on-fix ]
-    - id: ruff-format
+    - id: ruff-format
+
+- repo: local
+  hooks:
+    - id: ariadne-codegen
+      name: ariadne-codegen
+      entry: bash -c
+        echo "Running ariadne-codegen" && \
+        ariadne-codegen
+      language: system
+      types: [graphql]
+      pass_filenames: false

openhexa/cli/api.py

@@ -21,6 +21,7 @@
 from graphql.utilities import find_breaking_changes
 from jinja2 import Template
 
+from openhexa.cli.graphql.graphql_client import Client
 from openhexa.cli.settings import settings
 from openhexa.sdk.pipelines import get_local_workspace_config
 from openhexa.sdk.pipelines.runtime import get_pipeline
@@ -107,7 +108,16 @@ def get_library_versions() -> tuple[str, str]:
         return installed_version, installed_version
 
 
-def detect_graphql_breaking_changes(token):
+def _detect_graphql_breaking_changes_if_needed(token):
+    """Detect breaking changes if not done recently between the schema referenced in the SDK and the server using graphql-core."""
+    ONE_HOUR = 60 * 60
+    now_timestamp = int(datetime.now().timestamp())
+    if not settings.last_breaking_change_check or now_timestamp - settings.last_breaking_change_check > ONE_HOUR:
+        _detect_graphql_breaking_changes(token)
+        settings.last_breaking_change_check = now_timestamp
+
+
+def _detect_graphql_breaking_changes(token):
     """Detect breaking changes between the schema referenced in the SDK and the server using graphql-core."""
     stored_schema_obj = build_schema((Path(__file__).parent / "graphql" / "schema.generated.graphql").open().read())
     server_schema_obj = build_client_schema(
@@ -133,11 +143,7 @@ def detect_graphql_breaking_changes(token):
 
 def graphql(query: str, variables=None, token=None):
     """Check that there is no breaking change and perform a GraphQL request."""
-    ONE_HOUR = 60 * 60
-    now_timestamp = int(datetime.now().timestamp())
-    if not settings.last_breaking_change_check or now_timestamp - settings.last_breaking_change_check > ONE_HOUR:
-        detect_graphql_breaking_changes(token)
-        settings.last_breaking_change_check = now_timestamp
+    _detect_graphql_breaking_changes_if_needed(token)
     return _query_graphql(query, variables, token)
 
 
@@ -739,3 +745,47 @@ def is_dhis2_connection_up(workspace_slug: str, connection_slug: str) -> bool:
         },
     )
     return response["data"]["connectionBySlug"]["status"] == "UP"
+
+
+class OpenHexaClient(Client):
+    """OpenHexaClient is a class that provides methods to interact with the OpenHexa GraphQL API."""
+
+    def __init__(self, token=None):
+        """Initialize the OpenHexaClient with the OpenHexa API URL and headers."""
+        self._url = settings.api_url + "/graphql/"
+        self._token = token or settings.access_token
+
+        if not self._token:
+            raise InvalidTokenError("No token found for workspace")
+
+        super().__init__(
+            url=self._url,
+            headers={
+                "User-Agent": f"openhexa-cli/{version('openhexa.sdk')}",
+                "Authorization": f"Bearer {self._token}",
+            },
+        )
+        logging.getLogger("httpx").setLevel(
+            logging.WARNING
+        )  # HTTPX logs queries by default, we disable them here with WARNING level
+
+    def execute(self, query, **kwargs):
+        """Decorate parent execute method to log the GraphQL query and response."""
+        _detect_graphql_breaking_changes(token=self._token)
+
+        if settings.debug:
+            click.echo("")
+            click.echo("Graphql Query:")
+            click.echo(f"URL: {self.url}")
+            click.echo(f"Query: {query}")
+            variables = kwargs.get("variables", {})
+            click.echo(f"Variables: {variables}")
+
+        response = super().execute(query=query, **kwargs)
+
+        if settings.debug:
+            click.echo("")
+            click.echo("Graphql Response:")
+            click.echo(f"Response: {response}")
+
+        return response

openhexa/cli/cli.py

@@ -14,6 +14,7 @@
     DockerError,
     InvalidDefinitionError,
     NoActiveWorkspaceError,
+    OpenHexaClient,
     OutputDirectoryError,
     PipelineDirectoryError,
     create_pipeline,
@@ -24,7 +25,6 @@
     ensure_is_pipeline_dir,
     get_library_versions,
     get_pipeline_from_code,
-    get_pipelines,
     get_pipelines_pages,
     get_workspace,
     run_pipeline,
@@ -596,21 +596,19 @@ def pipelines_list():
     if settings.current_workspace is None:
         _terminate("No workspace activated", err=True)
 
-    workspace_pipelines = get_pipelines()
+    workspace_pipelines = (
+        OpenHexaClient().get_workspace_pipelines(workspace_slug=settings.current_workspace).pipelines.items
+    )
     if len(workspace_pipelines) == 0:
         click.echo(f"No pipelines in workspace {settings.current_workspace}")
         return
     click.echo("Pipelines:")
     for pipeline in workspace_pipelines:
-        if pipeline["type"] == "zipFile":
-            current_version = pipeline["currentVersion"].get("number") if pipeline["currentVersion"] else None
-            if current_version is not None:
-                current_version = f"v{current_version}"
-            else:
-                current_version = "N/A"
+        if pipeline.type == "zipFile":
+            current_version = f"v{pipeline.current_version.version_number}" if pipeline.current_version else "N/A"
         else:
             current_version = "Jupyter notebook"
-        click.echo(f"* {pipeline['code']} - {pipeline['name']} ({current_version})")
+        click.echo(f"* {pipeline.code} - {pipeline.name} ({current_version})")
 
 
 def _terminate(message: str, exception: Exception = None, err: bool = False):