Extract KQL Queries from Azure Workbooks (Workbook → App Tool)

{
  "items": [
    {
      "type": 1,
      "content": {
        "json": "## Azure Inventory Workbook\n### Change Log"
      }
    },
    {
      "type": 3,
      "content": {
        "query": "Resources | where type == 'microsoft.compute/virtualmachines'..."
      }
    },
    {
      "type": 9,
      "content": {
        "parameters": [...]
      }
    }
  ]
}

def flatten_items(workbook: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Extract all items from workbook JSON with type, name, and content."""
    items = []
    raw_items = workbook.get("items") or workbook.get("graph") or []

    if isinstance(raw_items, list):
        for i, it in enumerate(raw_items):
            t = it.get("type") or it.get("content",{}).get("type")
            name = it.get("name") or it.get("content",{}).get("name") or it.get("title")
            items.append({"index": i, "type": t, "name": name, "raw": it})

    # Handle ARM template format (value array)
    if not items and isinstance(workbook.get("value"), list):
        for entry in workbook["value"]:
            props = entry.get("properties", {})
            items2 = props.get("serializedData") or props.get("items")
            if isinstance(items2, str):
                try:
                    items2 = json.loads(items2)
                except Exception:
                    items2 = []
            if isinstance(items2, list):
                for i, it in enumerate(items2):
                    t = it.get("type") or it.get("content",{}).get("type")
                    name = it.get("name") or it.get("content",{}).get("name")
                    items.append({"index": i, "type": t, "name": name, "raw": it})

    return items

def extract_kql(item: Dict[str, Any]) -> Optional[str]:
    """Extract KQL query from item, handling multiple schema formats."""
    c = item.get("raw", {}).get("content", {})

    if isinstance(c, dict):
        # Direct query string
        if "query" in c and isinstance(c["query"], str):
            return c["query"]

        # Nested query object
        if "query" in c and isinstance(c["query"], dict) and "query" in c["query"]:
            return c["query"]["query"]

        # Versioned query format
        if "versionedQuery" in c and isinstance(c["versionedQuery"], dict):
            if "query" in c["versionedQuery"]:
                return c["versionedQuery"]["query"]

        # JSON-wrapped query
        if "json" in c and isinstance(c["json"], dict) and "query" in c["json"]:
            return c["json"]["query"]

    # Queries array (some workbooks)
    q = c.get("queries") if isinstance(c, dict) else None
    if isinstance(q, list) and q:
        first = q[0]
        if isinstance(first, dict) and "query" in first:
            return first["query"]

    return None

AttributeError: 'int' object has no attribute 'lower'

# Item type mappings (numeric → readable)
WORKBOOK_ITEM_TYPES = {
    1: "text",
    3: "query",
    9: "parameters",
    10: "links",
    11: "group",
    12: "metrics"
}

def normalize_type(raw_type: Any) -> str:
    """Convert workbook type (int or string) to normalized lowercase string."""
    if raw_type is None:
        return "unknown"
    if isinstance(raw_type, int):
        return WORKBOOK_ITEM_TYPES.get(raw_type, f"type{raw_type}")
    if isinstance(raw_type, str):
        return raw_type.lower()
    return str(raw_type).lower()

def get_type_display(raw_type: Any) -> str:
    """Get human-readable type name for display."""
    normalized = normalize_type(raw_type)
    if isinstance(raw_type, int):
        return f"{WORKBOOK_ITEM_TYPES.get(raw_type, f'type{raw_type}')} ({raw_type})"
    return normalized

import streamlit as st

st.set_page_config(page_title="Azure Workbook → App", layout="wide")
st.title("📒 Azure Monitor Workbook → App")

# File upload
uploaded = st.file_uploader("Upload an Azure Monitor Workbook JSON", type=["json"])

if not uploaded:
    st.info("Upload a workbook export (JSON).")
    st.stop()

# Parse workbook
workbook = load_workbook_json(uploaded.getvalue())
items = flatten_items(workbook)

st.success(f"Parsed {len(items)} item(s) from the workbook.")

# Sidebar navigation
with st.sidebar:
    st.markdown("### Items")
    def format_item(i):
        item = items[i]
        type_display = get_type_display(item.get('type'))
        name = item.get('name') or '(unnamed)'
        return f"{i}: {type_display} — {name}"

    idx = st.selectbox("Jump to item", list(range(len(items))), format_func=format_item)

# Display current item
current = items[idx]
itype = normalize_type(current.get('type'))

if "query" in itype or itype == "3":
    q = extract_kql(current)
    if q:
        st.code(q, language="kusto")
        # Optional: Execute query if workspace ID provided
        if run_queries and workspace_id:
            df = try_run_kql(workspace_id, q)
            if df is not None and not df.empty:
                st.dataframe(df.head(500), use_container_width=True)

# Extract all KQL queries
kql_rows = []
for it in items:
    k = extract_kql(it)
    if k:
        type_display = get_type_display(it.get('type'))
        kql_rows.append({
            "index": it["index"], 
            "type": type_display,
            "name": it.get("name") or "(unnamed)", 
            "query": k
        })

kql_df = pd.DataFrame(kql_rows)

# CSV export
st.download_button(
    "📥 Download all KQL to CSV", 
    kql_df.to_csv(index=False).encode("utf-8"), 
    "workbook_kql.csv", 
    "text/csv"
)

# Markdown export
md = io.StringIO()
md.write(f"# Workbook: {uploaded.name}\n\n")
md.write(f"**Items:** {len(items)}\n")
md.write(f"**KQL Queries:** {len(kql_rows)}\n\n")

for i, it in enumerate(items):
    type_display = get_type_display(it.get('type'))
    name = it.get('name') or '(unnamed)'
    md.write(f"## Item {i}: {type_display} — {name}\n\n")

    txt = extract_text(it)
    q = extract_kql(it)

    if txt:
        md.write(txt + "\n\n")
    if q:
        md.write("```kusto\n" + q + "\n```\n\n")

    md.write("---\n\n")

st.download_button(
    "📥 Download as Markdown", 
    md.getvalue().encode("utf-8"), 
    "workbook_export.md", 
    "text/markdown"
)

✅ Parsed 110 item(s) from the workbook.
✅ Found 41 KQL queries in workbook.

# Workbook: template.json

**Items:** 110
**KQL Queries:** 41

---

## Item 2: text (1) — Change Log

## Azure Inventory Workbook
### Change Log

|Version|Description|
|---|---|
|v1.1.1| Initial Release - October 2020 - BY
|v2.0.2 | Added VM state, fixed WVD Details - Feb 2022 - BY

---

## Item 9: query (3) — Azure Compute Summary

```kusto
Resources | where type == "microsoft.compute/virtualmachines"
| extend vmState = tostring(properties.extended.instanceView.powerState.displayStatus)
| extend vmState = iif(isempty(vmState), "VM State Unknown", (vmState))
| summarize count() by vmState

Resources 
| where type == "microsoft.compute/virtualmachines"
| extend vmID = tolower(id)
| extend osDiskId= tolower(tostring(properties.storageProfile.osDisk.managedDisk.id))
| join kind=leftouter(resources
    | where type =~ 'microsoft.compute/disks'
    | where properties !has 'Unattached'
    | project timeCreated, OS, osSku, osDiskSizeGB, osDiskId) on osDiskId
| join kind=leftouter(resources
    | where type =~ 'microsoft.compute/availabilitysets'
    | mv-expand VirtualMachine=properties.virtualMachines
    | project AvailabilitySetID = id, vmID, FaultDomainCount) on vmID
...

**CSV export:** All 41 queries with index, type, name, and full query text.

**Time to parse:** 10 seconds  
**Time to understand structure:** 5 minutes reading the Markdown  
**Time saved vs manual Portal clicking:** Hours

## What This Enabled

### Enhanced Billy York's Workbook

Original workbook: 50 Azure resource types  
Enhanced version: 200+ resource types  

**The tool let me:**
1. **Extract all existing queries** - see what resources were already tracked
2. **Identify patterns** - understand the query structure Billy used
3. **Add new resource types** - following the same patterns
4. **Test queries locally** - verify before deploying
5. **Document everything** - exported Markdown served as reference

[Read about the enhancement here →](/blog/modernizing-azure-workbooks/)

### Built Custom Workbooks

Using the extracted queries as templates:
- Created security-focused workbook (NSG rules, exposed endpoints)
- Built cost analysis workbook (unattached resources, stopped VMs)
- Developed compliance tracking workbook (tagging, policy violations)

**Pattern:** Extract queries from existing workbooks → modify → deploy new workbooks

### Migrated Workbooks to Code

For complex scenarios, converted workbook queries to Python scripts:
- Extracted all queries using this tool
- Rewrote as Python using Azure SDK
- Automated daily reports
- Integrated with alerting systems

## Optional: Execute Queries

The tool can actually **run the KQL queries** if you provide a Log Analytics workspace ID:

```python
def try_run_kql(workspace_id: str, query: str) -> Optional[pd.DataFrame]:
    """Execute KQL query against Log Analytics workspace."""
    if not AZURE_AVAILABLE or not workspace_id or not query:
        return None

    try:
        cred = DefaultAzureCredential(exclude_shared_token_cache_credential=True)
        client = LogsQueryClient(credential=cred)
        resp = client.query_workspace(workspace_id, query, timespan=None)

        if not resp.tables:
            return pd.DataFrame()

        t = resp.tables[0]
        cols = [c.name for c in t.columns]
        rows = [list(r) for r in t.rows]
        return pd.DataFrame(rows, columns=cols)
    except Exception as e:
        st.warning(f"KQL execution failed: {e}")
        return None

# Install
git clone https://github.com/yourusername/workbook_app
cd workbook_app
pip install -r requirements.txt

# Run locally
streamlit run streamlit_workbook_app.py

# Opens http://127.0.0.1:8502
# Upload your workbook JSON