# Copyright (C) 2020 Alteryx, Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Configurations for managing workspace."""
import json
import os
import shutil
import tempfile
import xml.etree.ElementTree as Et
from pathlib import Path
from typing import List, Optional
from ayx_plugin_sdk.cli.environment_helpers import download_pip_packages
from ayx_plugin_sdk.cli.workspace_tool import WorkspaceTool
from pydantic import BaseModel
import typer
import xmltodict
WORKSPACE_CONFIGURATION_FILE_NAME = "ayx_workspace.json"
REQUIREMENTS_FILE_NAME = "requirements.txt"
[docs]class Workspace(BaseModel):
"""Class wrapping all workspace configurations and handling basic workspace functions."""
workspace_dir: Path = Path(".")
tools: List[WorkspaceTool] = []
yxi_name: Optional[str] = None
designer_path: Optional[Path] = None
tool_family_name: Optional[str] = None
requirements_tool: Optional[str] = None
[docs] @classmethod
def setup_workspace_directory(cls, workspace_directory: Path) -> None:
"""Copy the base configs to the tool directory."""
cls._copy_base_workspace_config(workspace_directory)
cls._copy_base_requirements_file_to_workspace(workspace_directory)
workspace = cls(
workspace_dir=workspace_directory,
tools=[],
yxi_name=workspace_directory.resolve().name,
tool_family_name=workspace_directory.resolve().name,
)
workspace._write()
typer.echo(f"Created workspace directory: {workspace_directory.resolve().name}")
[docs] @classmethod
def initialize_workspace(cls, workspace_directory: Path) -> "Workspace":
"""Create a Workspace instance wrapping the configurations for the given workspace_directory."""
workspace_config_path = workspace_directory / WORKSPACE_CONFIGURATION_FILE_NAME
if not workspace_config_path.exists():
typer.echo(
"ERROR: Specified workspace_directory isn't a workspace directory, "
f"since a {WORKSPACE_CONFIGURATION_FILE_NAME} file doesn't exist."
"Please use an existing workspace, or specify a directory that doesn't exist yet."
)
raise typer.Exit(code=1)
with open(
str(workspace_directory / WORKSPACE_CONFIGURATION_FILE_NAME), "r"
) as f:
raw_config = json.loads(f.read())
workspace = cls(**raw_config)
workspace.workspace_dir = workspace_directory
for tool in workspace.tools:
tool.tool_directory = workspace_directory / tool.name
workspace._validate_tool_family_in_config_xml()
return workspace
def _write(self) -> None:
"""Write the workspace configurations to the workspace directory."""
self._write_config(
config_document=self.json(
exclude={
"workspace_dir": ...,
"tools": {"__all__": {"ToolDirectory"}},
},
indent=2,
),
workspace_directory=self.workspace_dir,
)
def _setup_workspace_tool(self, new_tool_name: str, example_tool_name: str) -> None:
"""Copy the example tool files to the tool folder."""
example_tool_directory = get_install_dir() / "examples" / example_tool_name
new_tool_directory = self.workspace_dir / new_tool_name
shutil.copytree(str(example_tool_directory), str(new_tool_directory))
self._update_config_file(example_tool_name, new_tool_name)
self._update_main_py(example_tool_name, new_tool_name)
@staticmethod
def _write_config(config_document: str, workspace_directory: Path) -> None:
workspace_config_path = workspace_directory / WORKSPACE_CONFIGURATION_FILE_NAME
with open(str(workspace_config_path), "w") as workspace_config_file:
workspace_config_file.write(config_document)
def _validate_tool_family_in_config_xml(self) -> None:
"""Ensure the workspace Config.xml and sub tools contains the tool family."""
config_xml_path = self.workspace_dir / "Config.xml"
self._set_tool_family_attribute(config_xml_path)
for tool in self.tools:
tool_config_path = tool.tool_directory / f"{tool.name}Config.xml"
self._set_tool_family_attribute(tool_config_path)
def _set_tool_family_attribute(self, config_xml_path: Path) -> None:
"""Set the ToolFamily attribute on a given Config.xml file."""
if not self.tool_family_name:
raise ValueError("Tool family name must be set.")
with open(str(config_xml_path), "r") as config_file:
tree = Et.parse(config_file)
root_node = tree.getroot()
engine_settings = root_node.find("EngineSettings")
if engine_settings is None:
raise ValueError("Config XML doesn't contain EngineSettings tag.")
engine_settings.attrib["ToolFamily"] = self.tool_family_name
tree.write(str(config_xml_path))
def _update_config_file(self, example_tool_name: str, new_tool_name: str) -> None:
old_config_path = (
self.workspace_dir / new_tool_name / f"{example_tool_name}Config.xml"
)
new_config_path = old_config_path.parent / f"{new_tool_name}Config.xml"
os.rename(str(old_config_path), str(new_config_path))
with open(str(new_config_path)) as f:
config = xmltodict.parse(f.read())
config["AlteryxJavaScriptPlugin"]["Properties"]["MetaInfo"][
"Name"
] = new_tool_name
with open(str(new_config_path), "w") as f:
f.write(xmltodict.unparse(config, pretty=True))
def _update_main_py(self, example_tool_name: str, new_tool_name: str) -> None:
"""Update the name of the tool in the main.py file."""
main_filepath = self.workspace_dir / new_tool_name / "main.py"
with open(str(main_filepath), "r") as f:
content = f.read()
content = content.replace(example_tool_name, new_tool_name)
with open(str(main_filepath), "w") as f:
f.write(content)
@staticmethod
def _copy_base_workspace_config(workspace_directory: Path) -> None:
install_dir = get_install_dir()
shutil.copytree(
str(install_dir / "assets" / "base_tool_config"), str(workspace_directory),
)
@staticmethod
def _copy_base_requirements_file_to_workspace(workspace_directory: Path) -> None:
install_dir = get_install_dir()
shutil.copy(
str(install_dir / "examples" / REQUIREMENTS_FILE_NAME),
str(workspace_directory),
)
[docs] def build_yxi(
self,
destination_dir: Path = Path("."),
name_override: Optional[str] = None,
include_dependencies: bool = True,
) -> Path:
"""Build a YXI for the workspace."""
yxi_name = name_override or self.yxi_name
if not yxi_name:
raise ValueError("No YXI name specified.")
if not self.requirements_tool:
raise ValueError(
"No requirements tool found. Have you added any tools to your workspace?"
)
if yxi_name != Path(yxi_name).stem:
typer.echo(
"ERROR: Detected that workspace YXI Name is a path. Please ensure this is a string"
" containing the base name of the YXI file (with no extension) to be generated in the current directory."
)
raise typer.Exit(code=1)
with tempfile.TemporaryDirectory() as yxi_temp_folder:
yxi_folder = create_workspace_yxi_folder(
self.workspace_dir, Path(yxi_temp_folder)
)
if include_dependencies:
temp_tools_requirements_path = yxi_folder / self.requirements_tool
copy_requirements_to_tool_and_link_wheels(
self.workspace_dir,
temp_tools_requirements_path,
self.requirements_tool,
)
tool_requirements_path = self.workspace_dir / REQUIREMENTS_FILE_NAME
download_pip_packages(
yxi_folder / self.requirements_tool / "wheels",
tool_requirements_path,
)
delete_pycache_directories(yxi_folder)
shutil.make_archive(yxi_name, "zip", str(yxi_folder))
archive_path = Path(f"{yxi_name}.zip")
dest_path = destination_dir / f"{yxi_name}.yxi"
shutil.move(str(archive_path), str(dest_path))
return dest_path
[docs]def get_install_dir() -> Path:
"""Get the current directory."""
return Path(__file__).parent.parent
[docs]def create_workspace_yxi_folder(
workspace_directory: Path,
temp_folder: Path,
tools_to_exclude: Optional[List[str]] = None,
) -> Path:
"""Create a temporary yxi folder."""
# The below statement fails with Path(temp_folder) / "tools
temp_tools_folder = os.path.join(str(temp_folder), "tools")
shutil.copytree(str(workspace_directory), str(temp_tools_folder))
if tools_to_exclude is not None:
for tool in tools_to_exclude:
shutil.rmtree(Path(temp_tools_folder) / tool)
return Path(temp_tools_folder)
[docs]def delete_pycache_directories(root_dir: Path) -> None:
"""Delete all the pycache subdirectories of a given root."""
pycache_dirs = [
Path(root) / directory
for root, directories, _ in os.walk(str(root_dir))
for directory in directories
if directory == "__pycache__"
]
for directory in pycache_dirs:
shutil.rmtree(str(directory), ignore_errors=True)