0527df717c
- modules.py: check /data/ volume first, then baked-in /bicep_modules_catalog.json - server.py: add POST /reload endpoint — reloads catalogs without restart - ilsp.nomad: add 'ilsp-data' host volume mounted at /data - Makefile: add push-catalogs, health-prod, run-with-data targets; DEVOPS_MCP_REPO var - scripts/push_catalogs.sh: SCP both catalogs to autobox + call /reload Workflow: sync scripts on Mac → make push-catalogs → completions live in <5s
91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
"""
|
|
LRU Bicep module catalog — loaded from the bundled catalog file at startup.
|
|
|
|
The catalog (bicep_modules_catalog.json) is baked into the Docker image at build time.
|
|
No runtime dependency on DevOpsMCP or any external service.
|
|
|
|
Provides completion items for LRU-internal Bicep modules with
|
|
higher sort priority than standard Azure modules.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import pathlib
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Catalog is baked into the image root at /bicep_modules_catalog.json
|
|
_CATALOG_PATHS = [
|
|
pathlib.Path("/data/bicep_modules_catalog.json"), # volume-mount (freshest)
|
|
pathlib.Path("/bicep_modules_catalog.json"), # baked into image (fallback)
|
|
pathlib.Path(__file__).parent.parent.parent / "bicep_modules_catalog.json", # dev
|
|
]
|
|
|
|
|
|
def _load_catalog() -> list[dict[str, Any]]:
|
|
"""Load modules from the bundled catalog file."""
|
|
for path in _CATALOG_PATHS:
|
|
if path.exists():
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
modules_raw = data.get("modules", {})
|
|
registry = data.get("registry", "iactemplatereg.azurecr.io")
|
|
modules = []
|
|
# modules is a dict: { "bicep/modules/appservice": { versions: [...], ... }, ... }
|
|
for mod_path, info in modules_raw.items():
|
|
versions = info.get("versions", ["latest"])
|
|
name = mod_path.split("/")[-1] if "/" in mod_path else mod_path
|
|
modules.append({
|
|
"name": name,
|
|
"path": mod_path,
|
|
"versions": versions,
|
|
"latest": versions[-1] if versions else "latest",
|
|
"registry": registry,
|
|
})
|
|
logger.info("Bicep catalog loaded from %s: %d modules", path, len(modules))
|
|
return modules
|
|
except Exception:
|
|
logger.exception("Failed to parse catalog at %s", path)
|
|
logger.warning("No bicep_modules_catalog.json found — completions disabled")
|
|
return []
|
|
|
|
|
|
class BicepModuleCatalog:
|
|
"""In-memory catalog of LRU Bicep modules, loaded once at startup."""
|
|
|
|
_modules: list[dict[str, Any]] = []
|
|
|
|
@classmethod
|
|
def load(cls) -> None:
|
|
"""Load catalog from disk. Call once at startup."""
|
|
cls._modules = _load_catalog()
|
|
|
|
@classmethod
|
|
def get_modules(cls) -> list[dict[str, Any]]:
|
|
return cls._modules
|
|
|
|
@classmethod
|
|
def as_completion_items(cls) -> list[dict[str, Any]]:
|
|
items = []
|
|
for mod in cls._modules:
|
|
ref = f"br/modules:{mod['path']}:{mod['latest']}"
|
|
items.append({
|
|
"label": mod["name"],
|
|
"kind": 9, # Module
|
|
"detail": f"LRU Bicep module — {mod['registry']}",
|
|
"insertText": ref,
|
|
"sortText": f"0_lru_{mod['name']}",
|
|
"documentation": {
|
|
"kind": "markdown",
|
|
"value": (
|
|
f"**{mod['name']}** (LRU internal)\n\n"
|
|
f"Registry: `{mod['registry']}`\n"
|
|
f"Versions: {', '.join(mod['versions'])}\n\n"
|
|
f"```bicep\nmodule {mod['name'].lower()} '{ref}' = {{\n"
|
|
f" name: '{mod['name'].lower()}'\n params: {{}}\n}}\n```"
|
|
),
|
|
},
|
|
})
|
|
return items
|