GraphQL-based quilt3.admin API

.github/workflows/test-quilt3-admin-codegen.yaml

@@ -0,0 +1,36 @@
+name: Test quilt3.admin code generation
+
+on:
+  push:
+    paths:
+      - '.github/workflows/test-quilt3-admin-codegen.yaml'
+      - 'shared/graphql/schema.graphql'
+      - 'api/python/quilt3-admin/**'
+      - 'api/python/quilt3/admin/_graphql_client/**'
+  pull_request:
+    paths:
+      - '.github/workflows/test-quilt3-admin-codegen.yaml'
+      - 'shared/graphql/schema.graphql'
+      - 'api/python/quilt3-admin/**'
+      - 'api/python/quilt3/admin/_graphql_client/**'
+  merge_group:
+
+jobs:
+  test-quilt3-admin-codegen:
+    name: test quilt3.admin generated code is up-to-date
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: ./api/python/quilt3-admin
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version-file: 'api/python/quilt3-admin/.python-version'
+          cache: 'pip'
+          cache-dependency-path: 'api/python/quilt3-admin/requirements.txt'
+      - run: pip install -r requirements.txt
+      - run: rm -r ../quilt3/admin/_graphql_client
+      - run: ariadne-codegen
+      - name: Check for changes
+        run: git diff --exit-code

api/python/.gitattributes

@@ -0,0 +1 @@
+quilt3/admin/_graphql_client/** linguist-generated

api/python/quilt3-admin/.python-version

@@ -0,0 +1 @@
+3.12

api/python/quilt3-admin/README.md

@@ -0,0 +1,10 @@
+# quilt3.admin GraphQL code generation
+
+```sh
+python -m venv venv
+python -m pip install -r requirements.txt
+ariadne-codegen
+```
+
+This will generate GraphQL client in `api/python/quilt3/admin/_graphql_client/` using
+GraphQL queries from `queries.graphql`.

api/python/quilt3-admin/base_client.py

@@ -0,0 +1,211 @@
+# This is
+# https://github.com/mirumee/ariadne-codegen/blob/5bfd63c5e7e3a8cc5293eb94deee638b7adab98d/ariadne_codegen/client_generators/dependencies/base_client.py
+# modified to use our requests session instead of httpx.
+# pylint: disable=relative-beyond-top-level
+import json
+from typing import IO, Any, Dict, List, Optional, Tuple, TypeVar, cast
+
+import requests
+from pydantic import BaseModel
+from pydantic_core import to_jsonable_python
+
+from quilt3 import session
+
+from .base_model import UNSET, Upload
+from .exceptions import (
+    GraphQLClientGraphQLMultiError,
+    GraphQLClientHttpError,
+    GraphQLClientInvalidResponseError,
+)
+
+Self = TypeVar("Self", bound="BaseClient")
+
+
+class BaseClient:
+    def __init__(
+        self,
+    ) -> None:
+        self.url = session.get_registry_url() + "/graphql"
+
+        self.http_client = session.get_session()
+
+    def __enter__(self: Self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: object,
+        exc_val: object,
+        exc_tb: object,
+    ) -> None:
+        self.http_client.close()
+
+    def execute(
+        self,
+        query: str,
+        operation_name: Optional[str] = None,
+        variables: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> requests.Response:
+        processed_variables, files, files_map = self._process_variables(variables)
+
+        if files and files_map:
+            return self._execute_multipart(
+                query=query,
+                operation_name=operation_name,
+                variables=processed_variables,
+                files=files,
+                files_map=files_map,
+                **kwargs,
+            )
+
+        return self._execute_json(
+            query=query,
+            operation_name=operation_name,
+            variables=processed_variables,
+            **kwargs,
+        )
+
+    def get_data(self, response: requests.Response) -> Dict[str, Any]:
+        if not 200 <= response.status_code < 300:
+            raise GraphQLClientHttpError(
+                status_code=response.status_code, response=response
+            )
+
+        try:
+            response_json = response.json()
+        except ValueError as exc:
+            raise GraphQLClientInvalidResponseError(response=response) from exc
+
+        if (not isinstance(response_json, dict)) or (
+            "data" not in response_json and "errors" not in response_json
+        ):
+            raise GraphQLClientInvalidResponseError(response=response)
+
+        data = response_json.get("data")
+        errors = response_json.get("errors")
+
+        if errors:
+            raise GraphQLClientGraphQLMultiError.from_errors_dicts(
+                errors_dicts=errors, data=data
+            )
+
+        return cast(Dict[str, Any], data)
+
+    def _process_variables(
+        self, variables: Optional[Dict[str, Any]]
+    ) -> Tuple[
+        Dict[str, Any], Dict[str, Tuple[str, IO[bytes], str]], Dict[str, List[str]]
+    ]:
+        if not variables:
+            return {}, {}, {}
+
+        serializable_variables = self._convert_dict_to_json_serializable(variables)
+        return self._get_files_from_variables(serializable_variables)
+
+    def _convert_dict_to_json_serializable(
+        self, dict_: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        return {
+            key: self._convert_value(value)
+            for key, value in dict_.items()
+            if value is not UNSET
+        }
+
+    def _convert_value(self, value: Any) -> Any:
+        if isinstance(value, BaseModel):
+            return value.model_dump(by_alias=True, exclude_unset=True)
+        if isinstance(value, list):
+            return [self._convert_value(item) for item in value]
+        return value
+
+    def _get_files_from_variables(
+        self, variables: Dict[str, Any]
+    ) -> Tuple[
+        Dict[str, Any], Dict[str, Tuple[str, IO[bytes], str]], Dict[str, List[str]]
+    ]:
+        files_map: Dict[str, List[str]] = {}
+        files_list: List[Upload] = []
+
+        def separate_files(path: str, obj: Any) -> Any:
+            if isinstance(obj, list):
+                nulled_list = []
+                for index, value in enumerate(obj):
+                    value = separate_files(f"{path}.{index}", value)
+                    nulled_list.append(value)
+                return nulled_list
+
+            if isinstance(obj, dict):
+                nulled_dict = {}
+                for key, value in obj.items():
+                    value = separate_files(f"{path}.{key}", value)
+                    nulled_dict[key] = value
+                return nulled_dict
+
+            if isinstance(obj, Upload):
+                if obj in files_list:
+                    file_index = files_list.index(obj)
+                    files_map[str(file_index)].append(path)
+                else:
+                    file_index = len(files_list)
+                    files_list.append(obj)
+                    files_map[str(file_index)] = [path]
+                return None
+
+            return obj
+
+        nulled_variables = separate_files("variables", variables)
+        files: Dict[str, Tuple[str, IO[bytes], str]] = {
+            str(i): (file_.filename, cast(IO[bytes], file_.content), file_.content_type)
+            for i, file_ in enumerate(files_list)
+        }
+        return nulled_variables, files, files_map
+
+    def _execute_multipart(
+        self,
+        query: str,
+        operation_name: Optional[str],
+        variables: Dict[str, Any],
+        files: Dict[str, Tuple[str, IO[bytes], str]],
+        files_map: Dict[str, List[str]],
+        **kwargs: Any,
+    ) -> requests.Response:
+        data = {
+            "operations": json.dumps(
+                {
+                    "query": query,
+                    "operationName": operation_name,
+                    "variables": variables,
+                },
+                default=to_jsonable_python,
+            ),
+            "map": json.dumps(files_map, default=to_jsonable_python),
+        }
+
+        return self.http_client.post(url=self.url, data=data, files=files, **kwargs)
+
+    def _execute_json(
+        self,
+        query: str,
+        operation_name: Optional[str],
+        variables: Dict[str, Any],
+        **kwargs: Any,
+    ) -> requests.Response:
+        headers: Dict[str, str] = {"Content-Type": "application/json"}
+        headers.update(kwargs.get("headers", {}))
+
+        merged_kwargs: Dict[str, Any] = kwargs.copy()
+        merged_kwargs["headers"] = headers
+
+        return self.http_client.post(
+            url=self.url,
+            data=json.dumps(
+                {
+                    "query": query,
+                    "operationName": operation_name,
+                    "variables": variables,
+                },
+                default=to_jsonable_python,
+            ),
+            **merged_kwargs,
+        )

api/python/quilt3-admin/exceptions.py

@@ -0,0 +1,87 @@
+# This is
+# https://github.com/mirumee/ariadne-codegen/blob/5bfd63c5e7e3a8cc5293eb94deee638b7adab98d/ariadne_codegen/client_generators/dependencies/exceptions.py
+# modified to use our requests instead of httpx.
+# pylint: disable=super-init-not-called
+from typing import Any, Dict, List, Optional, Union
+
+import requests
+
+
+class GraphQLClientError(Exception):
+    """Base exception."""
+
+
+class GraphQLClientHttpError(GraphQLClientError):
+    def __init__(self, status_code: int, response: requests.Response) -> None:
+        self.status_code = status_code
+        self.response = response
+
+    def __str__(self) -> str:
+        return f"HTTP status code: {self.status_code}"
+
+
+class GraphQLClientInvalidResponseError(GraphQLClientError):
+    def __init__(self, response: requests.Response) -> None:
+        self.response = response
+
+    def __str__(self) -> str:
+        return "Invalid response format."
+
+
+class GraphQLClientGraphQLError(GraphQLClientError):
+    def __init__(
+        self,
+        message: str,
+        locations: Optional[List[Dict[str, int]]] = None,
+        path: Optional[List[str]] = None,
+        extensions: Optional[Dict[str, object]] = None,
+        orginal: Optional[Dict[str, object]] = None,
+    ):
+        self.message = message
+        self.locations = locations
+        self.path = path
+        self.extensions = extensions
+        self.orginal = orginal
+
+    def __str__(self) -> str:
+        return self.message
+
+    @classmethod
+    def from_dict(cls, error: Dict[str, Any]) -> "GraphQLClientGraphQLError":
+        return cls(
+            message=error["message"],
+            locations=error.get("locations"),
+            path=error.get("path"),
+            extensions=error.get("extensions"),
+            orginal=error,
+        )
+
+
+class GraphQLClientGraphQLMultiError(GraphQLClientError):
+    def __init__(
+        self,
+        errors: List[GraphQLClientGraphQLError],
+        data: Optional[Dict[str, Any]] = None,
+    ):
+        self.errors = errors
+        self.data = data
+
+    def __str__(self) -> str:
+        return "; ".join(str(e) for e in self.errors)
+
+    @classmethod
+    def from_errors_dicts(
+        cls, errors_dicts: List[Dict[str, Any]], data: Optional[Dict[str, Any]] = None
+    ) -> "GraphQLClientGraphQLMultiError":
+        return cls(
+            errors=[GraphQLClientGraphQLError.from_dict(e) for e in errors_dicts],
+            data=data,
+        )
+
+
+class GraphQLClientInvalidMessageFormat(GraphQLClientError):
+    def __init__(self, message: Union[str, bytes]) -> None:
+        self.message = message
+
+    def __str__(self) -> str:
+        return "Invalid message format."

api/python/quilt3-admin/pyproject.toml

@@ -0,0 +1,21 @@
+[tool.ariadne-codegen]
+schema_path = "../../../shared/graphql/schema.graphql"
+queries_path = "queries.graphql"
+target_package_path = "../quilt3/admin/"
+target_package_name = "_graphql_client"
+files_to_include = [
+    "exceptions.py",
+]
+async_client = false
+base_client_file_path = "base_client.py"
+base_client_name = "BaseClient"
+include_all_inputs = false
+include_all_enums = false
+plugins = [
+    "ariadne_codegen.contrib.shorter_results.ShorterResultsPlugin",
+    "ariadne_codegen.contrib.shorter_results.ShorterResultsPlugin",
+    "ariadne_codegen.contrib.shorter_results.ShorterResultsPlugin",
+]
+
+[tool.ariadne-codegen.scalars.Datetime]
+type = "datetime.datetime"