Rob
05/06/2023, 3:59 AMkedro viz
over Host '0.0.0.0'
? I want to show them expanded by default. Here's an example of the behavior I'm seeing: https://brawlstars-retention-pipeline-6u27jcczha-uw.a.run.app/, and this doesn't happen with localhost.
Can someone suggest a way to modify the code used in this thread to show the nodes expanded? Thanks 🙂"""`kedro_viz.server` provides utilities to launch a webserver for Kedro pipeline visualisation."""
import webbrowser
from pathlib import Path
from typing import Any, Dict, Optional
import uvicorn
from fastapi.encoders import jsonable_encoder
from <http://kedro.io|kedro.io> import DataCatalog
from kedro.pipeline import Pipeline
from watchgod import run_process
import os
from kedro_viz.api import apps
from kedro_viz.api.rest.responses import EnhancedORJSONResponse, get_default_response
from kedro_viz.constants import DEFAULT_HOST, DEFAULT_PORT
from kedro_viz.data_access import DataAccessManager, data_access_manager
from kedro_viz.database import create_db_engine
from kedro_viz.integrations.kedro import data_loader as kedro_data_loader
from kedro_viz.models.experiment_tracking import Base
DEV_PORT = 4142
def is_localhost(host) -> bool:
"""Check whether a host is a localhost"""
return host in ("127.0.0.1", "localhost", "0.0.0.0")
def populate_data(
data_access_manager: DataAccessManager,
catalog: DataCatalog,
pipelines: Dict[str, Pipeline],
session_store_location: Optional[Path],
): # pylint: disable=redefined-outer-name
"""Populate data repositories. Should be called once on application start
if creatinge an api app from project.
"""
if session_store_location:
database_engine, session_class = create_db_engine(session_store_location)
Base.metadata.create_all(bind=database_engine)
data_access_manager.set_db_session(session_class)
data_access_manager.add_catalog(catalog)
data_access_manager.add_pipelines(pipelines)
def run_server(
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
browser: Optional[bool] = None,
load_file: Optional[str] = None,
save_file: Optional[str] = None,
pipeline_name: Optional[str] = None,
env: Optional[str] = None,
project_path: Optional[str] = None,
autoreload: bool = False,
extra_params: Optional[Dict[str, Any]] = None,
): # pylint: disable=redefined-outer-name, too-many-locals
"""Run an uvicorn server with a FastAPI app that either launches API response data
from a file or from reading data from a real Kedro project.
Args:
host: the host to launch the webserver
port: the port to launch the webserver
browser: whether to open the default browser automatically
load_file: if a valid JSON containing API response data is provided,
the API of the server is created from the JSON.
save_file: if provided, the data returned by the API will be saved to a file.
pipeline_name: the optional name of the pipeline to visualise.
env: the optional environment of the pipeline to visualise.
If not provided, it will use Kedro's default, which is "local".
autoreload: Whether the API app should support autoreload.
project_path: the optional path of the Kedro project that contains the pipelines
to visualise. If not supplied, the current working directory will be used.
extra_params: Optional dictionary containing extra project parameters
for underlying KedroContext. If specified, will update (and therefore
take precedence over) the parameters retrieved from the project
configuration.
"""
if load_file is None:
path = Path(project_path) if project_path else Path.cwd()
catalog, pipelines, session_store_location = kedro_data_loader.load_data(
path, env, extra_params
)
pipelines = (
pipelines
if pipeline_name is None
else {pipeline_name: pipelines[pipeline_name]}
)
populate_data(data_access_manager, catalog, pipelines, session_store_location)
if save_file:
default_response = get_default_response()
jsonable_default_response = jsonable_encoder(default_response)
encoded_default_response = EnhancedORJSONResponse.encode_to_human_readable(
jsonable_default_response
)
Path(save_file).write_bytes(encoded_default_response)
app = apps.create_api_app_from_project(path, autoreload)
else:
app = apps.create_api_app_from_file(load_file)
if browser and is_localhost(host):
webbrowser.open_new(f"http://{host}:{port}/")
uvicorn.run(app, host=host, port=port, log_config=None)
if __name__ == "__main__": # pragma: no cover
import argparse
from kedro.framework.startup import bootstrap_project
port = int(os.environ.get("PORT", 4141))
project_path = os.getcwd()
bootstrap_project(project_path)
run_process(
project_path,
run_server,
kwargs={
"host": "0.0.0.0",
"port": port,
"project_path": str(project_path),
},
)