Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce get_kedro_project_json_data function for VSCode integration #2049

Merged
merged 10 commits into from
Sep 3, 2024
30 changes: 28 additions & 2 deletions package/kedro_viz/api/rest/responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

# pylint: disable=missing-class-docstring,invalid-name
import abc
import json
import logging
from importlib.metadata import PackageNotFoundError
from typing import Any, Dict, List, Optional, Union
Expand Down Expand Up @@ -400,17 +401,42 @@ def get_package_compatibilities_response(
return package_requirements_response


def write_api_response_to_fs(file_path: str, response: Any, remote_fs: Any):
"""Encodes, enhances responses and writes it to a file"""
def get_encoded_response(response: Any) -> bytes:
"""Encodes and enhances the default response using human-readable format."""
jsonable_response = jsonable_encoder(response)
encoded_response = EnhancedORJSONResponse.encode_to_human_readable(
jsonable_response
)

return encoded_response


def write_api_response_to_fs(file_path: str, response: Any, remote_fs: Any):
"""Get encoded responses and writes it to a file"""
encoded_response = get_encoded_response(response)

with remote_fs.open(file_path, "wb") as file:
file.write(encoded_response)


def get_kedro_project_json_data():
astrojuanlu marked this conversation as resolved.
Show resolved Hide resolved
"""Decodes the default response and returns the Kedro project JSON data.
This will be used in VSCode extension to get current Kedro project data."""
encoded_response = get_encoded_response(get_default_response())

try:
response_str = encoded_response.decode("utf-8")
json_data = json.loads(response_str)
except UnicodeDecodeError as exc: # pragma: no cover
json_data = None
logger.error("Failed to decode response string. Error: %s", str(exc))
except json.JSONDecodeError as exc: # pragma: no cover
json_data = None
logger.error("Failed to parse JSON data. Error: %s", str(exc))

return json_data


def save_api_main_response_to_fs(main_path: str, remote_fs: Any):
"""Saves API /main response to a directory."""
try:
Expand Down
23 changes: 23 additions & 0 deletions package/tests/test_api/test_rest/test_responses.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# pylint: disable=too-many-lines
import json
import logging
import operator
from pathlib import Path
Expand All @@ -13,6 +14,7 @@
from kedro_viz.api.rest.responses import (
EnhancedORJSONResponse,
PackageCompatibilityAPIResponse,
get_kedro_project_json_data,
get_package_compatibilities_response,
save_api_main_response_to_fs,
save_api_node_response_to_fs,
Expand Down Expand Up @@ -964,6 +966,27 @@ def test_write_api_response_to_fs(
mockremote_fs.open.assert_called_once_with(file_path, "wb")
mock_encode_to_human_readable.assert_called_once()

def test_get_kedro_project_json_data(self, mocker):
expected_json_data = {"key": "value"}
encoded_response = json.dumps(expected_json_data).encode("utf-8")

mock_get_default_response = mocker.patch(
"kedro_viz.api.rest.responses.get_default_response",
return_value={"key": "value"},
)
mock_get_encoded_response = mocker.patch(
"kedro_viz.api.rest.responses.get_encoded_response",
return_value=encoded_response,
)

json_data = get_kedro_project_json_data()

mock_get_default_response.assert_called_once()
mock_get_encoded_response.assert_called_once_with(
mock_get_default_response.return_value
)
assert json_data == expected_json_data

def test_save_api_main_response_to_fs(self, mocker):
expected_default_response = {"test": "json"}
main_path = "/main"
Expand Down
Loading