fix(openhexaClient): circular dependency

openhexa/__init__.py

@@ -0,0 +1 @@
+"""OpenHexa package initialization."""

openhexa/cli/api.py

@@ -8,24 +8,19 @@
 import os
 import tempfile
 import typing
-from datetime import datetime
-from importlib.metadata import version
 from pathlib import Path
 from zipfile import ZipFile
 
 import click
 import docker
-import requests
 from docker.models.containers import Container
-from graphql import build_client_schema, build_schema, get_introspection_query
-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.graphql.openhexa_client import graphql
 from openhexa.sdk.pipelines import get_local_workspace_config
 from openhexa.sdk.pipelines.runtime import get_pipeline
-from openhexa.utils import create_requests_session, stringcase
+from openhexa.utils import stringcase
 
 
 class InvalidDefinitionError(Exception):
@@ -46,24 +41,6 @@ class OutputDirectoryError(Exception):
     pass
 
 
-class APIError(Exception):
-    """Raised when an error occurs while interacting with the API."""
-
-    pass
-
-
-class InvalidTokenError(APIError):
-    """Raised when the token is invalid."""
-
-    pass
-
-
-class GraphQLError(APIError):
-    """Raised when a GraphQL request returns an error."""
-
-    pass
-
-
 class PipelineDirectoryError(Exception):
     """Raised when the pipeline directory is not a directory or does not exist."""
 
@@ -90,108 +67,6 @@ class PermissionDenied(Exception):
     pass
 
 
-def get_library_versions() -> tuple[str, str]:
-    """Return the current version and the one on PyPi."""
-    # Get the currently installed version
-    installed_version = version("openhexa.sdk")
-
-    # Get the latest version available on PyPI
-    try:
-        response = requests.get("https://pypi.org/pypi/openhexa.sdk/json")
-        latest_version = response.json()["info"]["version"]
-        return installed_version, latest_version
-    except requests.RequestException:
-        logging.error(
-            "Could not check for the latest version of the openhexa.sdk package.",
-            exc_info=True,
-        )
-        return installed_version, installed_version
-
-
-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(
-        _query_graphql(get_introspection_query(input_value_deprecation=True), token=token)
-    )
-
-    breaking_changes = find_breaking_changes(stored_schema_obj, server_schema_obj)
-    if breaking_changes:
-        current_version, latest_version = get_library_versions()
-        click.secho(
-            f"⚠️ Breaking changes detected between the SDK (version {current_version}) and the server:",
-            fg="red",
-        )
-        for change in breaking_changes:
-            click.secho(f"- {change.description}", fg="yellow")
-        click.secho(
-            "This could lead to unexpected results.\n"
-            f"Please update the SDK to the latest version {latest_version} "
-            f"(using `pip install openhexa-sdk=={latest_version}`) or use a version of the SDK compatible with the server.",
-            fg="red",
-        )
-
-
-def graphql(query: str, variables=None, token=None):
-    """Check that there is no breaking change and perform a GraphQL request."""
-    _detect_graphql_breaking_changes_if_needed(token)
-    return _query_graphql(query, variables, token)
-
-
-def _query_graphql(query: str, variables=None, token=None):
-    """Perform a GraphQL request."""
-    url = settings.api_url + "/graphql/"
-    if token is None:
-        token = settings.access_token
-
-    if token is None:
-        raise InvalidTokenError("No token found for workspace")
-
-    if settings.debug:
-        click.echo("")
-        click.echo("Graphql Query:")
-        click.echo(f"URL: {url}")
-        click.echo(f"Query: {query}")
-        click.echo(f"Variables: {variables}")
-
-    session = create_requests_session()
-
-    response = session.post(
-        url,
-        headers={
-            "User-Agent": f"openhexa-cli/{version('openhexa.sdk')}",
-            "Authorization": f"Bearer {token}",
-        },
-        json={"query": query, "variables": variables},
-    )
-    try:
-        response.raise_for_status()
-    except requests.exceptions.HTTPError as e:
-        raise GraphQLError(str(e))
-
-    data = response.json()
-
-    if settings.debug:
-        click.echo("Graphql Response:")
-        click.echo(data)
-        click.echo("")
-
-    if data.get("errors"):
-        if data.get("errors")[0].get("extensions", {}).get("code") == "UNAUTHENTICATED":
-            raise InvalidTokenError
-        raise GraphQLError(data["errors"])
-    return data["data"]
-
-
 def get_skeleton_dir():
     """Get the path to the skeleton directory."""
     return Path(__file__).parent / "skeleton"
@@ -745,47 +620,3 @@ 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,7 +14,6 @@
     DockerError,
     InvalidDefinitionError,
     NoActiveWorkspaceError,
-    OpenHexaClient,
     OutputDirectoryError,
     PipelineDirectoryError,
     create_pipeline,
@@ -23,14 +22,14 @@
     delete_pipeline,
     download_pipeline_sourcecode,
     ensure_is_pipeline_dir,
-    get_library_versions,
     get_pipeline_from_code,
     get_pipelines_pages,
     get_workspace,
     run_pipeline,
     upload_pipeline,
 )
 from openhexa.cli.settings import settings, setup_logging
+from openhexa.graphql.openhexa_client import OpenHexaClient, get_library_versions
 from openhexa.sdk.pipelines.exceptions import PipelineNotFound
 from openhexa.sdk.pipelines.runtime import get_pipeline
 
@@ -597,7 +596,9 @@ def pipelines_list():
         _terminate("No workspace activated", err=True)
 
     workspace_pipelines = (
-        OpenHexaClient().get_workspace_pipelines(workspace_slug=settings.current_workspace).pipelines.items
+        OpenHexaClient(settings.api_url, settings.access_token)
+        .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}")

openhexa/cli/settings.py

@@ -6,8 +6,6 @@
 
 import click
 
-from openhexa.sdk.pipelines.log_level import LogLevel
-
 CONFIGFILE_PATH = os.path.expanduser("~") + "/.openhexa.ini"
 
 
@@ -78,11 +76,6 @@ def workspaces(self):
         """Return the workspaces from the settings file."""
         return self._file_config["workspaces"]
 
-    @property
-    def log_level(self) -> LogLevel:
-        """Return the log level from the environment variables."""
-        return LogLevel.parse_log_level(os.getenv("HEXA_LOG_LEVEL"))
-
     def activate(self, workspace: str):
         """Set the current workspace in the settings file."""
         if workspace not in self.workspaces:

openhexa/graphql/__init__.py

@@ -0,0 +1 @@
+"""OpenHexa-SDK GraphQL client to communicate with OpenHexa API."""