Source code for scaffolder.generator
#!/usr/bin/env python3
"""
scaffolder/generator.py - now only supports 'widget' kind
2025-06-22 (fixed header/variable injection per-operation)
"""
import shutil
import textwrap
from datetime import date
from pathlib import Path
from scaffolder.utils import (
_build_readme,
_make_default_cfg,
_make_schema,
_prepare_mutation,
_prepare_query,
_prepare_subscription,
_repl,
_write,
ensure_dir_permissions,
ensure_file_permissions,
js_lit,
pascal_to_camel,
pascal_to_kebab,
)
from scaffolder.widget_generator import generate_widget
ROOT = Path(__file__).resolve().parent.parent
TEMPLATE_DIR = ROOT / "templates"
OUT_DIR = ROOT / "output"
TODAY = date.today()
YEAR = TODAY.year
NL = "\n"
MAKE_SUBMODULE_PATH = ".make"
MAKE_SUBMODULE_URL = (
"https://gitlab.com/ska-telescope/sdi/ska-cicd-makefile.git"
)
DEFAULT_FEATURE_FLAGS = {
"workspace_switch": False,
"variable_selector": False,
"starter_table": False,
}
def _coerce_default_for_type(value: object, gql_type: str | None) -> object:
type_name = str(gql_type or "").strip()
if type_name.startswith("["):
return value
if type_name.rstrip("!") in {"String", "ID"} and value is not None:
text = str(value).strip()
if len(text) >= 2 and text[0] == "'" and text[-1] == "'":
text = text[1:-1]
return text
return value
def _to_tango_attribute_defaults(
full_names: object, endpoint: object | None
) -> list[object]:
names = full_names if isinstance(full_names, list) else []
endpoint_str = endpoint if isinstance(endpoint, str) else ""
out: list[object] = []
for item in names:
if not isinstance(item, str):
continue
value = item.strip()
if not value:
continue
if endpoint_str:
out.append({"endpoint": endpoint_str, "attribute": value})
else:
out.append(value)
return out
[docs]
def normalize_feature_flags(raw: dict | None) -> dict[str, bool]:
merged = dict(DEFAULT_FEATURE_FLAGS)
if isinstance(raw, dict):
for key in merged:
merged[key] = bool(raw.get(key))
return merged
[docs]
def build_feature_schema_parts(
flags: dict[str, bool],
) -> tuple[dict[str, object], dict[str, str], dict[str, str], str]:
extra_defaults: dict[str, object] = {}
extra_schema: dict[str, str] = {}
extra_props: dict[str, str] = {}
sdk_schema_imports: set[str] = {
"INPUT_FORMATTED_SCHEMA_TYPE",
"INPUT_FORMATTED_PRESET_DESCRIPTIONS_TOKEN",
}
extra_defaults["title"] = ""
extra_schema["title"] = (
"{ type: INPUT_FORMATTED_SCHEMA_TYPE, title: 'Title', "
"description: `Title string can use: ${INPUT_FORMATTED_PRESET_DESCRIPTIONS_TOKEN}`, "
"default: '' }"
)
extra_props["title"] = "string"
if flags["workspace_switch"]:
extra_defaults["workspaceTarget"] = ""
extra_schema["workspaceTarget"] = (
"{ type: WORKSPACES, title: 'Workspace target', description: 'Workspace to switch to when this example action is triggered.', default: '' }"
)
extra_props["workspaceTarget"] = "string"
sdk_schema_imports.add("WORKSPACES")
if flags["variable_selector"]:
extra_defaults["selectedVariableKey"] = ""
extra_defaults["selectedVariableValue"] = ""
extra_schema["selectedVariableKey"] = (
"{ type: VARIABLES, title: 'Variable key', description: 'Dashboard variable key used by the generated example control.', default: '' }"
)
extra_schema["selectedVariableValue"] = (
"{ type: 'string', title: 'Variable value', description: 'Value written into the selected dashboard variable by the example control.', default: '' }"
)
extra_props["selectedVariableKey"] = "string"
extra_props["selectedVariableValue"] = "string"
sdk_schema_imports.add("VARIABLES")
sdk_imports_literal = ", ".join(sorted(sdk_schema_imports))
return extra_defaults, extra_schema, extra_props, sdk_imports_literal
[docs]
def scaffold_widget(
pascal: str,
kind: str,
*,
operations: list,
var_defaults: dict,
var_types: dict,
app_id: str,
feature_flags: dict | None = None,
):
pascal = to_pascal(pascal)
if kind != "widget":
yield f"ERROR: Unknown kind '{kind}'."
return
kebab = pascal_to_kebab(pascal)
camel = pascal_to_camel(pascal)
repo_slug = f"ska-octopus-{kebab}-widget"
flags = normalize_feature_flags(feature_flags)
has_poll = any(
(op.get("type") or "").lower() == "polling" for op in operations
)
has_stream = any(
(op.get("type") or "").lower() == "stream" for op in operations
)
config_var_defaults = dict(var_defaults)
config_var_types = dict(var_types)
widget_var_defaults = dict(var_defaults)
widget_var_types = dict(var_types)
for key, default in list(config_var_defaults.items()):
config_var_defaults[key] = _coerce_default_for_type(
default, config_var_types.get(key)
)
widget_options: dict[str, object] = {}
(
extra_defaults,
extra_schema,
extra_props,
sdk_schema_imports,
) = build_feature_schema_parts(flags)
uses_tango_attribute_pair = (
"fullNames" in config_var_defaults or "fullNames" in config_var_types
)
if uses_tango_attribute_pair:
endpoint_default = config_var_defaults.get("endpoint")
attr_defaults = _to_tango_attribute_defaults(
config_var_defaults.get("fullNames"), endpoint_default
)
config_var_defaults.pop("fullNames", None)
config_var_types.pop("fullNames", None)
config_var_defaults.pop("endpoint", None)
config_var_types.pop("endpoint", None)
extra_defaults["attributes"] = attr_defaults
extra_schema["attributes"] = (
"{ type: INPUT_TANGO_ATTRIBUTES, title: 'Tango attributes', "
"description: 'Select one or more Tango attributes that will be normalized into the GraphQL fullNames variable at runtime.', "
f"default: {js_lit(attr_defaults)} }}"
)
extra_props["attributes"] = (
"Array<{ endpoint?: string; attribute?: string } | string>"
)
imports = {part for part in sdk_schema_imports.split(", ") if part}
imports.add("INPUT_TANGO_ATTRIBUTES")
sdk_schema_imports = ", ".join(sorted(imports))
widget_options["tango_attributes"] = {
"config_key": "attributes",
"full_names_var": "fullNames",
"endpoint_var": "endpoint",
}
widget_options["optional_vars"] = ["fullNames", "endpoint"]
widget_var_defaults.setdefault("fullNames", [])
widget_var_defaults.setdefault("endpoint", "")
widget_var_types.setdefault("fullNames", "[String!]")
widget_var_types.setdefault("endpoint", "String")
if has_poll and has_stream:
extra_defaults["useLiveData"] = True
extra_schema["useLiveData"] = (
"{ type: 'boolean', title: 'Use live data', description: 'Enable stream updates when available. Disable to use polling on the dashboard refresh cadence.', default: true }"
)
extra_props["useLiveData"] = "boolean"
widget_options["live_data_toggle"] = True
OUT_DIR.mkdir(parents=True, exist_ok=True)
ensure_dir_permissions(OUT_DIR)
dest = OUT_DIR / repo_slug
shutil.rmtree(dest, ignore_errors=True)
ops_lines = ["import { gql } from '@apollo/client';", ""]
poll_arr, stream_arr, mut_arr = [], [], []
def _gql_doc(body: str) -> str:
return "gql`\n" + textwrap.indent(body.strip(), " ") + "\n `"
def _emit_docs(name: str, docs: list[str]) -> list[str]:
if not docs:
return [f"export const {name} = [];"]
return [
f"export const {name} = [",
" " + ",\n ".join(docs),
"];",
]
for op in operations:
otype = op.get("type", "").lower()
body_raw = op.get("gql", "")
op_var_types = {
k: v.get("type") for k, v in (op.get("vars") or {}).items()
}
if otype == "polling":
body = _prepare_query(body_raw, op_var_types)
poll_arr.append(_gql_doc(body))
elif otype == "stream":
body = _prepare_subscription(body_raw, op_var_types)
stream_arr.append(_gql_doc(body))
elif otype == "mutation":
body = _prepare_mutation(body_raw, op_var_types)
mut_arr.append(_gql_doc(body))
else:
yield f"ERROR: Unknown op type '{otype}'"
return
ops_lines += _emit_docs("pollingDocs", poll_arr)
ops_lines.append("")
ops_lines += _emit_docs("streamDocs", stream_arr)
ops_lines.append("")
ops_lines += _emit_docs("mutationDocs", mut_arr)
ops_lines.append("")
_write(dest / "src/graphql/ops.ts", NL.join(ops_lines))
yield "Wrote file: src/graphql/ops.ts"
generate_widget(
pascal,
camel,
widget_var_defaults,
widget_var_types,
operations,
dest,
feature_flags=flags,
extra_props=extra_props,
widget_options=widget_options,
)
yield f"Wrote file: src/{pascal}Widget.tsx"
var_names = sorted(config_var_defaults)
default_cfg_lit = _make_default_cfg(
var_names, config_var_defaults, extra_defaults
)
schema_literal = _make_schema(var_names, config_var_types, extra_schema)
combined_names = sorted(
[*config_var_defaults.keys(), *extra_defaults.keys()]
)
combined_defaults = {**config_var_defaults, **extra_defaults}
combined_types = {
**config_var_types,
**{key: "SDK custom type" for key in extra_schema},
}
ph = {
"PASCAL": pascal,
"CAMEL": camel,
"KEBAB": kebab,
"YEAR": str(YEAR),
"DEFAULT_CFG": default_cfg_lit,
"SCHEMA": schema_literal,
"APP_ID": app_id,
"GITLAB_USER": "ska-octopus",
"GITLAB_REPO": repo_slug,
"SDK_SCHEMA_IMPORTS": sdk_schema_imports,
"SDK_SCHEMA_IMPORTS_LINE": (
f"import {{ {sdk_schema_imports} }} from '@ska-octopus-widget-sdk/widget-sdk';"
if sdk_schema_imports
else ""
),
}
files = {
"package.json": TEMPLATE_DIR / "package.json",
"tsup.config.ts": TEMPLATE_DIR / "tsup.config.ts",
"vite.config.ts": TEMPLATE_DIR / "vite.config.ts",
"vite.config.lib.ts": TEMPLATE_DIR / "vite.config.lib.ts",
"tsconfig.json": TEMPLATE_DIR / "tsconfig.json",
"tsconfig.build.json": TEMPLATE_DIR / "tsconfig.build.json",
".gitignore": TEMPLATE_DIR / "gitignore",
".gitmodules": TEMPLATE_DIR / "gitmodules",
".npmrc": TEMPLATE_DIR / "npmrc",
"Makefile": TEMPLATE_DIR / "Makefile",
"CHANGELOG.md": TEMPLATE_DIR / "CHANGELOG.md",
"eslint.config.js": TEMPLATE_DIR / "eslint.config.js",
".prettierrc": TEMPLATE_DIR / "prettierrc",
".prettierignore": TEMPLATE_DIR / "prettierignore",
".husky/pre-commit": TEMPLATE_DIR / "husky.pre-commit",
"src/css.d.ts": TEMPLATE_DIR / "src/css.d.ts",
".gitlab-ci.yml": TEMPLATE_DIR / "gitlab-ci.yml",
"vitest.config.ts": TEMPLATE_DIR / "vitest.config.ts",
".readthedocs.yml": TEMPLATE_DIR / "readthedocs.yml",
"docs/requirements-docs.txt": TEMPLATE_DIR
/ "docs/requirements-docs.txt",
"docs/src/conf.py": TEMPLATE_DIR / "docs/conf.py",
"docs/src/index.md": TEMPLATE_DIR / "docs/index.md",
"docs/src/getting-started.md": TEMPLATE_DIR
/ "docs/getting-started.md",
"docs/src/architecture.md": TEMPLATE_DIR / "docs/architecture.md",
"docs/src/configuration.md": TEMPLATE_DIR / "docs/configuration.md",
"docs/src/widget-functionality.md": TEMPLATE_DIR
/ "docs/widget-functionality.md",
"docs/src/changelog.md": TEMPLATE_DIR / "docs/changelog.md",
"docs/src/coverage.md": TEMPLATE_DIR / "docs/coverage.md",
".release": TEMPLATE_DIR / "release",
"scripts/generate_changelog.py": TEMPLATE_DIR
/ "scripts/generate_changelog.py",
"dev/index.html": TEMPLATE_DIR / "dev/index.html",
"dev/main.tsx": TEMPLATE_DIR / "dev/main.widget.tsx",
"src/index.ts": TEMPLATE_DIR / "src/index.widget.ts",
f"src/{pascal}Widget.module.css": TEMPLATE_DIR
/ "src/widget.module.css",
"src/smoke.test.tsx": TEMPLATE_DIR / "src/smoke.test.tsx",
"tests/setup.tsx": TEMPLATE_DIR / "tests/setup.tsx",
"README.md": None,
}
for rel, tpl in files.items():
raw = (
_build_readme(
pascal,
repo_slug,
combined_names,
combined_defaults,
combined_types,
operations,
)
if rel == "README.md"
else tpl.read_text(encoding="utf-8")
)
rendered = _repl(raw, ph)
_write(dest / rel, rendered)
yield f"Wrote file: {rel}"
yield (
"INFO: configure .make as a git submodule (do not vendor files): "
f"'git submodule add {MAKE_SUBMODULE_URL} {MAKE_SUBMODULE_PATH}'"
)
zip_path = Path(
shutil.make_archive(str(OUT_DIR / repo_slug), "zip", root_dir=dest)
)
ensure_file_permissions(zip_path)
yield f"ZIP_READY:{zip_path}"
yield "Done."