Shareable Kedro-viz backend implementation

.gitignore

@@ -65,3 +65,4 @@ coverage.xml
 
 # Kedro
 *.log
+

package/features/steps/lower_requirements.txt

@@ -6,6 +6,7 @@ aiofiles==22.1.0
 uvicorn[standard]==0.22.0
 watchgod==0.8.2
 plotly==4.0
+packaging==23.0
 pandas==1.3; python_version < '3.10'
 pandas==1.5; python_version >= '3.10'
 sqlalchemy==1.4

package/kedro_viz/api/apps.py

@@ -68,7 +68,7 @@ def create_api_app_from_project(
         # frontend e2e tests via Cypress
         app.mount("/static", StaticFiles(directory=_HTML_DIR / "static"), name="static")
 
-    # everytime the server reloads, a new app with a new timestamp will be created.
+    # every time the server reloads, a new app with a new timestamp will be created.
     # this is used as an etag embedded in the frontend for client to use when making requests.
     app_etag = _create_etag()
 
@@ -131,6 +131,24 @@ async def index():
 
     @app.get("/api/main", response_class=JSONResponse)
     async def main():
-        return json.loads(Path(filepath).read_text(encoding="utf8"))
+        return json.loads((Path(filepath) / "main").read_text(encoding="utf8"))
+
+    @app.get("/api/nodes/{node_id}", response_class=JSONResponse)
+    async def get_node_metadata(node_id):
+        return json.loads(  # pragma: no cover
+            (Path(filepath) / "nodes" / node_id).read_text(encoding="utf8")
+        )
+
+    @app.get("/api/pipelines/{pipeline_id}", response_class=JSONResponse)
+    async def get_registered_pipeline(pipeline_id):
+        return json.loads(  # pragma: no cover
+            (Path(filepath) / "pipelines" / pipeline_id).read_text(encoding="utf8")
+        )
+
+    @app.get("/api/deploy-viz-metadata", response_class=JSONResponse)
+    async def get_deployed_viz_metadata():
+        return json.loads(  # pragma: no cover
+            (Path(filepath) / "deploy-viz-metadata").read_text(encoding="utf8")
+        )
 
     return app

package/kedro_viz/api/rest/requests.py

@@ -0,0 +1,9 @@
+"""`kedro_viz.api.rest.requests` defines REST request types."""
+from pydantic import BaseModel
+
+
+class S3DeployerConfiguration(BaseModel):
+    """Credentials for S3 Deployer."""
+
+    region: str
+    bucket_name: str

package/kedro_viz/api/rest/responses.py

@@ -1,13 +1,33 @@
 """`kedro_viz.api.rest.responses` defines REST response types."""
 # pylint: disable=missing-class-docstring,too-few-public-methods,invalid-name
 import abc
+import logging
 from typing import Any, Dict, List, Optional, Union
 
+import fsspec
 import orjson
-from fastapi.responses import ORJSONResponse
+import packaging
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse, ORJSONResponse
+from kedro.io.core import get_protocol_and_path
 from pydantic import BaseModel
 
+from kedro_viz.api.rest.utils import get_package_version
 from kedro_viz.data_access import data_access_manager
+from kedro_viz.models.flowchart import (
+    DataNode,
+    DataNodeMetadata,
+    ParametersNodeMetadata,
+    TaskNode,
+    TaskNodeMetadata,
+    TranscodedDataNode,
+    TranscodedDataNodeMetadata,
+)
+
+logger = logging.getLogger(__name__)
+
+_FSSPEC_PACKAGE_NAME = "fsspec"
+_FSSPEC_COMPATIBLE_VERSION = "2023.9.0"
 
 
 class APIErrorMessage(BaseModel):
@@ -244,6 +264,39 @@ class GraphAPIResponse(BaseAPIResponse):
     selected_pipeline: str
 
 
+class PackageCompatibilityAPIResponse(BaseAPIResponse):
+    package_name: str
+    package_version: str
+    is_compatible: bool
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "package_name": "fsspec",
+                "package_version": "2023.9.1",
+                "is_compatible": True,
+            }
+        }
+
+
+class EnhancedORJSONResponse(ORJSONResponse):
+    @staticmethod
+    def encode_to_human_readable(content: Any) -> bytes:
+        """A method to encode the given content to JSON, with the
+        proper formatting to write a human-readable file.
+
+        Returns:
+            A bytes object containing the JSON to write.
+
+        """
+        return orjson.dumps(
+            content,
+            option=orjson.OPT_INDENT_2
+            | orjson.OPT_NON_STR_KEYS
+            | orjson.OPT_SERIALIZE_NUMPY,
+        )
+
+
 def get_default_response() -> GraphAPIResponse:
     """Default response for `/api/main`."""
     default_selected_pipeline_id = (
@@ -273,19 +326,149 @@ def get_default_response() -> GraphAPIResponse:
     )
 
 
-class EnhancedORJSONResponse(ORJSONResponse):
-    @staticmethod
-    def encode_to_human_readable(content: Any) -> bytes:
-        """A method to encode the given content to JSON, with the
-        proper formatting to write a human-readable file.
+def get_node_metadata_response(node_id: str):
+    """API response for `/api/nodes/node_id`."""
+    node = data_access_manager.nodes.get_node_by_id(node_id)
+    if not node:
+        return JSONResponse(status_code=404, content={"message": "Invalid node ID"})
 
-        Returns:
-            A bytes object containing the JSON to write.
+    if not node.has_metadata():
+        return JSONResponse(content={})
 
-        """
-        return orjson.dumps(
-            content,
-            option=orjson.OPT_INDENT_2
-            | orjson.OPT_NON_STR_KEYS
-            | orjson.OPT_SERIALIZE_NUMPY,
+    if isinstance(node, TaskNode):
+        return TaskNodeMetadata(node)
+
+    if isinstance(node, DataNode):
+        return DataNodeMetadata(node)
+
+    if isinstance(node, TranscodedDataNode):
+        return TranscodedDataNodeMetadata(node)
+
+    return ParametersNodeMetadata(node)
+
+
+def get_selected_pipeline_response(registered_pipeline_id: str):
+    """API response for `/api/pipeline/pipeline_id`."""
+    if not data_access_manager.registered_pipelines.has_pipeline(
+        registered_pipeline_id
+    ):
+        return JSONResponse(status_code=404, content={"message": "Invalid pipeline ID"})
+
+    modular_pipelines_tree = (
+        data_access_manager.create_modular_pipelines_tree_for_registered_pipeline(
+            registered_pipeline_id
+        )
+    )
+
+    return GraphAPIResponse(
+        nodes=data_access_manager.get_nodes_for_registered_pipeline(  # type: ignore
+            registered_pipeline_id
+        ),
+        edges=data_access_manager.get_edges_for_registered_pipeline(  # type: ignore
+            registered_pipeline_id
+        ),
+        tags=data_access_manager.tags.as_list(),
+        layers=data_access_manager.get_sorted_layers_for_registered_pipeline(
+            registered_pipeline_id
+        ),
+        pipelines=data_access_manager.registered_pipelines.as_list(),
+        selected_pipeline=registered_pipeline_id,
+        modular_pipelines=modular_pipelines_tree,  # type: ignore
+    )
+
+
+def get_package_compatibilities_response():
+    """API response for `/api/package_compatibility`."""
+    package_name = _FSSPEC_PACKAGE_NAME
+    package_version = get_package_version(package_name)
+    is_compatible = packaging.version.parse(package_version) >= packaging.version.parse(
+        _FSSPEC_COMPATIBLE_VERSION
+    )
+    return PackageCompatibilityAPIResponse(
+        package_name=package_name,
+        package_version=package_version,
+        is_compatible=is_compatible,
+    )
+
+
+def write_api_response_to_fs(file_path: str, response: Any, remote_fs: Any):
+    """Encodes, enhances responses and writes it to a file"""
+    jsonable_response = jsonable_encoder(response)
+    encoded_response = EnhancedORJSONResponse.encode_to_human_readable(
+        jsonable_response
+    )
+
+    with remote_fs.open(file_path, "wb") as file:
+        file.write(encoded_response)
+
+
+def save_api_main_response_to_fs(main_loc: str, remote_fs: Any):
+    """Saves API /main response to a file."""
+    try:
+        write_api_response_to_fs(main_loc, get_default_response(), remote_fs)
+    except Exception as exc:  # pragma: no cover
+        logger.exception("Failed to save default response. Error: %s", str(exc))
+        raise exc
+
+
+def save_api_node_response_to_fs(nodes_loc: str, remote_fs: Any):
+    """Saves API /nodes/{node} response to a file."""
+    for nodeId in data_access_manager.nodes.get_node_ids():
+        try:
+            write_api_response_to_fs(
+                f"{nodes_loc}/{nodeId}", get_node_metadata_response(nodeId), remote_fs
+            )
+        except Exception as exc:  # pragma: no cover
+            logger.exception(
+                "Failed to save node data for node ID %s. Error: %s", nodeId, str(exc)
+            )
+            raise exc
+
+
+def save_api_pipeline_response_to_fs(pipelines_loc: str, remote_fs: Any):
+    """Saves API /pipelines/{pipeline} response to a file."""
+    for pipelineId in data_access_manager.registered_pipelines.get_pipeline_ids():
+        try:
+            write_api_response_to_fs(
+                f"{pipelines_loc}/{pipelineId}",
+                get_selected_pipeline_response(pipelineId),
+                remote_fs,
+            )
+        except Exception as exc:  # pragma: no cover
+            logger.exception(
+                "Failed to save pipeline data for pipeline ID %s. Error: %s",
+                pipelineId,
+                str(exc),
+            )
+            raise exc
+
+
+def save_api_responses_to_fs(filepath: str):
+    """Saves all Kedro Viz API responses to a file."""
+    try:
+        protocol, path = get_protocol_and_path(filepath)
+        remote_fs = fsspec.filesystem(protocol)
+
+        logger.debug(
+            """Saving/Uploading api files to %s""",
+            filepath,
+        )
+
+        main_loc = f"{path}/api/main"
+        nodes_loc = f"{path}/api/nodes"
+        pipelines_loc = f"{path}/api/pipelines"
+
+        if protocol == "file":
+            remote_fs.makedirs(path, exist_ok=True)
+            remote_fs.makedirs(nodes_loc, exist_ok=True)
+            remote_fs.makedirs(pipelines_loc, exist_ok=True)
+
+        save_api_main_response_to_fs(main_loc, remote_fs)
+        save_api_node_response_to_fs(nodes_loc, remote_fs)
+        save_api_pipeline_response_to_fs(pipelines_loc, remote_fs)
+
+    except Exception as exc:  # pragma: no cover
+        logger.exception(
+            "An error occurred while preparing data for saving. Error: %s", str(exc)
         )
+        raise exc