#!/usr/bin/env python3
"""
octopus_widget_scaffolder.py - CLI entry point for the widget scaffolder.
"""
from __future__ import annotations
import argparse
import json
import sys
from collections.abc import Iterable
from pathlib import Path
from typing import Any
from scaffolder.generator import scaffold_widget
Operation = dict[str, Any]
VarSpec = dict[str, dict[str, Any]]
[docs]
def to_pascal(name: str) -> str:
return name[0].upper() + name[1:] if name else name
def _load_vars(raw: str | None) -> VarSpec:
if not raw:
return {}
data = json.loads(raw)
if not isinstance(data, dict):
raise ValueError("vars-json must be a JSON object")
return {k: dict(v or {}) for k, v in data.items()}
def _load_ops(raw: str) -> list[Operation]:
data = json.loads(raw)
if isinstance(data, dict) and "operations" in data:
data = data["operations"]
if not isinstance(data, list):
raise ValueError("ops-json must decode to a list of operations")
out: list[Operation] = []
for item in data:
if not isinstance(item, dict):
raise ValueError("Each operation must be a JSON object")
if "type" not in item or "gql" not in item:
raise ValueError("Operation requires 'type' and 'gql' keys")
copy = dict(item)
if "vars" in copy and copy["vars"] is not None:
if not isinstance(copy["vars"], dict):
raise ValueError("Operation 'vars' must be a JSON object")
copy["vars"] = {k: dict(v or {}) for k, v in copy["vars"].items()}
out.append(copy)
return out
def _collapse_vars(operations: Iterable[Operation]) -> tuple[dict, dict]:
defaults: dict[str, Any] = {}
types: dict[str, Any] = {}
for op in operations:
for key, spec in (op.get("vars") or {}).items():
defaults[key] = spec.get("default")
types[key] = spec.get("type")
return defaults, types
def _build_legacy_operations(
kind: str, gql: str, var_spec: VarSpec
) -> list[Operation]:
if kind == "basic":
return []
kinds: list[str]
if kind == "all":
kinds = ["polling", "stream", "mutation"]
else:
kinds = [kind]
operations: list[Operation] = []
for op_type in kinds:
operations.append(
{
"type": op_type,
"gql": gql,
"vars": {k: dict(v) for k, v in var_spec.items()} or None,
}
)
return operations
[docs]
def main(argv: list[str] | None = None) -> None:
p = argparse.ArgumentParser(description="Octopus widget scaffolder (CLI)")
p.add_argument("--name", required=True, help="Widget name in PascalCase")
p.add_argument(
"--kind",
required=True,
choices=["basic", "polling", "stream", "mutation", "all", "mix"],
help="Widget type",
)
p.add_argument("--gql-file", help="Path to file with GraphQL selection")
p.add_argument("--gql", help="GraphQL selection string")
p.add_argument(
"--vars-json",
help='Variables JSON, e.g. {"VAR":{"default":123,"type":"Int!"}}',
)
p.add_argument(
"--ops-json", help="Full operations JSON (list or {operations: [...]})"
)
p.add_argument("--app-id", required=True, help="GitLab project ID")
p.add_argument(
"--with-workspace-switch",
action="store_true",
help="Add workspace-switch example wiring to generated widget",
)
p.add_argument(
"--with-variable-selector",
action="store_true",
help="Add variable selector + variable update example wiring",
)
p.add_argument(
"--with-custom-input-examples",
action="store_true",
help=argparse.SUPPRESS,
)
p.add_argument(
"--with-starter-table",
action="store_true",
help="Add starter interactive table scaffold",
)
args = p.parse_args(argv)
name = to_pascal(args.name)
try:
var_spec = _load_vars(args.vars_json)
except ValueError as exc:
print(f"ERROR: invalid vars-json: {exc}", file=sys.stderr)
sys.exit(1)
operations: list[Operation] = []
if args.kind == "mix" and not args.ops_json:
print("ERROR: --ops-json is required when --kind mix", file=sys.stderr)
sys.exit(1)
if args.ops_json:
try:
operations = _load_ops(args.ops_json)
except ValueError as exc:
print(f"ERROR: invalid ops-json: {exc}", file=sys.stderr)
sys.exit(1)
else:
if args.kind != "basic":
source = ""
if args.gql_file:
try:
source = (
Path(args.gql_file).read_text(encoding="utf-8").strip()
)
except (
Exception
) as exc: # pragma: no cover - surfaced via stderr
print(
f"ERROR: could not read --gql-file: {exc}",
file=sys.stderr,
)
sys.exit(1)
else:
source = (args.gql or "").strip()
if not source:
print("ERROR: provide --gql-file or --gql", file=sys.stderr)
sys.exit(1)
else:
source = ""
operations = _build_legacy_operations(args.kind, source, var_spec)
# Backfill vars for operations that do not specify them explicitly
if var_spec:
for op in operations:
op.setdefault("vars", {k: dict(v) for k, v in var_spec.items()})
var_defaults, var_types = _collapse_vars(operations)
feature_flags = {
"workspace_switch": bool(args.with_workspace_switch),
"variable_selector": bool(args.with_variable_selector),
"starter_table": bool(args.with_starter_table),
}
for message in scaffold_widget(
name,
"widget",
operations=operations,
var_defaults=var_defaults,
var_types=var_types,
app_id=args.app_id,
feature_flags=feature_flags,
):
print(message)
if __name__ == "__main__":
main()