feat: bake bicep catalog into image; fix dict-based modules parsing
Some checks failed
Build and Deploy iLSP / build-and-deploy (push) Has been cancelled
Build and Deploy iLSP / test (push) Successful in 18s

- Remove all DevOpsMCP/aiohttp runtime deps from BicepModuleCatalog
- BicepModuleCatalog.load() reads bicep_modules_catalog.json from disk at startup (sync)
- Fix _load_catalog: catalog uses dict {path: {versions, schema}} not a list
- server.py: call BicepModuleCatalog.load() synchronously, not via asyncio.gather
- Dockerfile: COPY bicep_modules_catalog.json into both builder + runtime stages
- Health endpoint now reports bicep_modules: 27

Verified locally: make run-quick → health returns pypi_packages:40 bicep_modules:27
This commit is contained in:
Henrik Jess Nielsen
2026-05-10 13:40:48 +02:00
parent 44da791f5a
commit 6b38cbd70c
4 changed files with 12658 additions and 95 deletions

View File

@@ -1,84 +1,69 @@
"""
LRU Bicep module catalog — fetched from DevOpsMCP at startup.
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 asyncio
import json
import logging
import os
import time
import pathlib
from typing import Any
import aiohttp
logger = logging.getLogger(__name__)
DEVOPS_MCP_URL = os.getenv("DEVOPS_MCP_URL", "https://devops-mcp.i80.dk")
REFRESH_INTERVAL = 3600 # 1 hour
# Catalog is baked into the image root at /bicep_modules_catalog.json
_CATALOG_PATHS = [
pathlib.Path("/bicep_modules_catalog.json"),
pathlib.Path(__file__).parent.parent.parent / "bicep_modules_catalog.json",
]
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:
"""Fetches and caches LRU Bicep modules from DevOpsMCP."""
"""In-memory catalog of LRU Bicep modules, loaded once at startup."""
_modules: list[dict[str, Any]] = []
_last_refresh: float = 0
_lock = asyncio.Lock()
@classmethod
async def get_modules(cls) -> list[dict[str, Any]]:
if not cls._modules or time.time() - cls._last_refresh > REFRESH_INTERVAL:
await cls._refresh()
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
async def start_background_refresh(cls) -> None:
asyncio.create_task(cls._refresh_loop())
@classmethod
async def _refresh_loop(cls) -> None:
while True:
await cls._refresh()
await asyncio.sleep(REFRESH_INTERVAL)
@classmethod
async def _refresh(cls) -> None:
async with cls._lock:
try:
modules = await cls._fetch_from_devops_mcp()
cls._modules = modules
cls._last_refresh = time.time()
logger.info("Bicep module catalog refreshed: %d modules", len(modules))
except Exception:
logger.exception("Failed to refresh Bicep module catalog — using stale cache")
@classmethod
async def _fetch_from_devops_mcp(cls) -> list[dict[str, Any]]:
"""Call DevOpsMCP /api/bicep-modules REST endpoint."""
url = f"{DEVOPS_MCP_URL}/api/bicep-modules"
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) as session:
async with session.get(url) as resp:
if resp.status != 200:
raise RuntimeError(f"DevOpsMCP returned {resp.status}")
data = await resp.json()
modules = []
for m in data.get("modules", []):
path = m.get("path", "")
versions = m.get("versions", ["latest"])
name = path.split("/")[-1] if "/" in path else path
modules.append({
"name": name,
"path": path,
"versions": versions,
"latest": versions[-1] if versions else "latest",
"registry": data.get("registry", "iactemplatereg.azurecr.io"),
})
return modules
@classmethod
def as_completion_items(cls) -> list[dict[str, Any]]:
items = []
@@ -89,7 +74,7 @@ class BicepModuleCatalog:
"kind": 9, # Module
"detail": f"LRU Bicep module — {mod['registry']}",
"insertText": ref,
"sortText": f"0_lru_{mod['name']}", # sorts above standard az modules
"sortText": f"0_lru_{mod['name']}",
"documentation": {
"kind": "markdown",
"value": (