235 lines
8.9 KiB
Python
235 lines
8.9 KiB
Python
|
"""
|
||
|
|
PipelineTemplateCatalog — in-memory catalog of AzDO pipeline templates and
|
||
|
|
GitHub Actions reusable workflows.
|
||
|
|
|
||
|
|
Loaded from pipeline_templates_catalog.json (baked into image or volume-mounted).
|
||
|
|
Provides LSP completion items for template keys, parameter names, and allowed values.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import pathlib
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
_CATALOG_PATHS = [
|
||
|
|
pathlib.Path("/data/pipeline_templates_catalog.json"), # volume-mount (freshest)
|
||
|
|
pathlib.Path("/pipeline_templates_catalog.json"), # baked into image
|
||
|
|
pathlib.Path(__file__).parent.parent.parent / "pipeline_templates_catalog.json", # dev
|
||
|
|
]
|
||
|
|
|
||
|
|
# LSP completion item kinds
|
||
|
|
_KIND_MODULE = 9
|
||
|
|
_KIND_VALUE = 12
|
||
|
|
_KIND_PROPERTY = 10
|
||
|
|
|
||
|
|
|
||
|
|
def _load_catalog() -> dict[str, dict[str, Any]]:
|
||
|
|
for path in _CATALOG_PATHS:
|
||
|
|
if path.exists():
|
||
|
|
try:
|
||
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
||
|
|
templates = data.get("templates", {})
|
||
|
|
logger.info(
|
||
|
|
"Pipeline template catalog loaded from %s: %d templates", path, len(templates)
|
||
|
|
)
|
||
|
|
return templates
|
||
|
|
except Exception:
|
||
|
|
logger.exception("Failed to parse pipeline template catalog at %s", path)
|
||
|
|
logger.info("No pipeline_templates_catalog.json found — template completions disabled")
|
||
|
|
return {}
|
||
|
|
|
||
|
|
|
||
|
|
class PipelineTemplateCatalog:
|
||
|
|
"""In-memory catalog of pipeline templates, loaded once at startup."""
|
||
|
|
|
||
|
|
_templates: dict[str, dict[str, Any]] = {}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def load(cls) -> None:
|
||
|
|
"""Load catalog from disk. Call once at startup (or on /reload)."""
|
||
|
|
cls._templates = _load_catalog()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_template(cls, key: str) -> dict[str, Any] | None:
|
||
|
|
return cls._templates.get(key)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def all_keys(cls) -> list[str]:
|
||
|
|
return list(cls._templates.keys())
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def template_count(cls) -> int:
|
||
|
|
return len(cls._templates)
|
||
|
|
|
||
|
|
# ── AzDO completions ─────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def azdo_template_completion_items(cls) -> list[dict[str, Any]]:
|
||
|
|
"""Completion items for AzDO template paths (shown when typing a template: value)."""
|
||
|
|
items = []
|
||
|
|
for key, tmpl in cls._templates.items():
|
||
|
|
if tmpl.get("format") != "azdo":
|
||
|
|
continue
|
||
|
|
alias = tmpl.get("alias", "pipeline-templates")
|
||
|
|
path = tmpl.get("path", key)
|
||
|
|
nparams = len(tmpl.get("parameters", []))
|
||
|
|
items.append({
|
||
|
|
"label": path,
|
||
|
|
"kind": _KIND_MODULE,
|
||
|
|
"detail": f"@{alias} — {nparams} params",
|
||
|
|
"insertText": path,
|
||
|
|
"sortText": f"0_lru_{path}",
|
||
|
|
"documentation": _azdo_template_doc(path, alias, tmpl),
|
||
|
|
})
|
||
|
|
return items
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def azdo_param_completion_items(cls, template_key: str) -> list[dict[str, Any]]:
|
||
|
|
"""Completion items for AzDO parameter names inside a parameters: block."""
|
||
|
|
tmpl = cls._templates.get(template_key)
|
||
|
|
if not tmpl:
|
||
|
|
return []
|
||
|
|
items = []
|
||
|
|
for i, p in enumerate(tmpl.get("parameters", [])):
|
||
|
|
name = p["name"]
|
||
|
|
required = p.get("required", False)
|
||
|
|
label = name + ("*" if required else "")
|
||
|
|
detail_parts = [p.get("type", "string")]
|
||
|
|
if required:
|
||
|
|
detail_parts.append("required")
|
||
|
|
if "default" in p:
|
||
|
|
detail_parts.append(f"default: {p['default']!r}")
|
||
|
|
items.append({
|
||
|
|
"label": label,
|
||
|
|
"filterText": name,
|
||
|
|
"kind": _KIND_PROPERTY,
|
||
|
|
"detail": " · ".join(detail_parts),
|
||
|
|
"insertText": f"{name}: ",
|
||
|
|
"sortText": f"0_{i:03d}_{name}",
|
||
|
|
"documentation": _param_doc(p),
|
||
|
|
})
|
||
|
|
return items
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def azdo_param_value_items(cls, template_key: str, param_name: str) -> list[dict[str, Any]]:
|
||
|
|
"""Completion items for allowed values of an AzDO parameter."""
|
||
|
|
tmpl = cls._templates.get(template_key)
|
||
|
|
if not tmpl:
|
||
|
|
return []
|
||
|
|
for p in tmpl.get("parameters", []):
|
||
|
|
if p["name"] == param_name:
|
||
|
|
return [
|
||
|
|
{
|
||
|
|
"label": str(v),
|
||
|
|
"kind": _KIND_VALUE,
|
||
|
|
"insertText": str(v),
|
||
|
|
"sortText": f"0_{i:03d}_{v}",
|
||
|
|
}
|
||
|
|
for i, v in enumerate(p.get("allowed", []))
|
||
|
|
]
|
||
|
|
return []
|
||
|
|
|
||
|
|
# ── GHA completions ───────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def gha_workflow_completion_items(cls) -> list[dict[str, Any]]:
|
||
|
|
"""Completion items for GHA reusable workflow references (uses: value)."""
|
||
|
|
items = []
|
||
|
|
for key, tmpl in cls._templates.items():
|
||
|
|
if tmpl.get("format") != "gha":
|
||
|
|
continue
|
||
|
|
org = tmpl.get("org", "")
|
||
|
|
repo = tmpl.get("repo", "")
|
||
|
|
path = tmpl.get("path", "")
|
||
|
|
ref = tmpl.get("ref", "main")
|
||
|
|
full_ref = f"{org}/{repo}/{path}@{ref}"
|
||
|
|
nparams = len(tmpl.get("parameters", []))
|
||
|
|
items.append({
|
||
|
|
"label": full_ref,
|
||
|
|
"kind": _KIND_MODULE,
|
||
|
|
"detail": f"{org}/{repo} — {nparams} inputs",
|
||
|
|
"insertText": full_ref,
|
||
|
|
"sortText": f"0_lru_{repo}_{path}",
|
||
|
|
"documentation": _gha_workflow_doc(full_ref, tmpl),
|
||
|
|
})
|
||
|
|
return items
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def gha_input_completion_items(cls, template_key: str) -> list[dict[str, Any]]:
|
||
|
|
"""Completion items for GHA workflow_call input names inside a with: block."""
|
||
|
|
tmpl = cls._templates.get(template_key)
|
||
|
|
if not tmpl:
|
||
|
|
return []
|
||
|
|
items = []
|
||
|
|
for i, p in enumerate(tmpl.get("parameters", [])):
|
||
|
|
name = p["name"]
|
||
|
|
required = p.get("required", False)
|
||
|
|
label = name + ("*" if required else "")
|
||
|
|
detail_parts = [p.get("type", "string")]
|
||
|
|
if required:
|
||
|
|
detail_parts.append("required")
|
||
|
|
if "default" in p:
|
||
|
|
detail_parts.append(f"default: {p['default']!r}")
|
||
|
|
items.append({
|
||
|
|
"label": label,
|
||
|
|
"filterText": name,
|
||
|
|
"kind": _KIND_PROPERTY,
|
||
|
|
"detail": " · ".join(detail_parts),
|
||
|
|
"insertText": f"{name}: ",
|
||
|
|
"sortText": f"0_{i:03d}_{name}",
|
||
|
|
"documentation": _param_doc(p),
|
||
|
|
})
|
||
|
|
return items
|
||
|
|
|
||
|
|
|
||
|
|
# ── Documentation helpers ─────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
def _param_doc(p: dict[str, Any]) -> dict[str, str]:
|
||
|
|
lines = [f"**`{p['name']}`** ({p.get('type', 'string')})"]
|
||
|
|
if p.get("required"):
|
||
|
|
lines.append("_Required_")
|
||
|
|
if "default" in p:
|
||
|
|
lines.append(f"Default: `{p['default']!r}`")
|
||
|
|
if p.get("description"):
|
||
|
|
lines.append(f"\n{p['description']}")
|
||
|
|
allowed = p.get("allowed", [])
|
||
|
|
if allowed:
|
||
|
|
lines.append("\nAllowed: " + " | ".join(f"`{v}`" for v in allowed))
|
||
|
|
return {"kind": "markdown", "value": "\n\n".join(lines)}
|
||
|
|
|
||
|
|
|
||
|
|
def _azdo_template_doc(path: str, alias: str, tmpl: dict[str, Any]) -> dict[str, str]:
|
||
|
|
params = tmpl.get("parameters", [])
|
||
|
|
required = [p["name"] for p in params if p.get("required")]
|
||
|
|
optional = [p["name"] for p in params if not p.get("required")]
|
||
|
|
lines = [
|
||
|
|
f"**AzDO template** — `{path}@{alias}`",
|
||
|
|
"",
|
||
|
|
f"Reference: `- template: {path}@{alias}`",
|
||
|
|
"",
|
||
|
|
]
|
||
|
|
if required:
|
||
|
|
lines.append("**Required params:** " + ", ".join(f"`{n}`" for n in required))
|
||
|
|
if optional:
|
||
|
|
lines.append("**Optional params:** " + ", ".join(f"`{n}`" for n in optional))
|
||
|
|
return {"kind": "markdown", "value": "\n".join(lines)}
|
||
|
|
|
||
|
|
|
||
|
|
def _gha_workflow_doc(full_ref: str, tmpl: dict[str, Any]) -> dict[str, str]:
|
||
|
|
params = tmpl.get("parameters", [])
|
||
|
|
required = [p["name"] for p in params if p.get("required")]
|
||
|
|
optional = [p["name"] for p in params if not p.get("required")]
|
||
|
|
lines = [
|
||
|
|
f"**GHA reusable workflow** — `{full_ref}`",
|
||
|
|
"",
|
||
|
|
f"Reference: `uses: {full_ref}`",
|
||
|
|
"",
|
||
|
|
]
|
||
|
|
if required:
|
||
|
|
lines.append("**Required inputs:** " + ", ".join(f"`{n}`" for n in required))
|
||
|
|
if optional:
|
||
|
|
lines.append("**Optional inputs:** " + ", ".join(f"`{n}`" for n in optional))
|
||
|
|
return {"kind": "markdown", "value": "\n".join(lines)}
|