Sync a Postman collection with an OpenAPI specification **without** overwriting team‑specific tweaks (auth, test scripts, headers, etc.)

postman_sync.py

"""
postman_sync.py
────────────────
Sync a Postman collection with an OpenAPI specification **without** overwriting
team‑specific tweaks (auth, test scripts, headers, etc.).

What this minimal script does
─────────────────────────────
1.  Loads an OpenAPI (JSON or YAML) file.
2.  Sanitises it (replaces enum examples with <enum> to avoid noisy diffs).
3.  Imports the spec into Postman via the public API → creates a **temporary**
    collection.
4.  Downloads that generated collection, deletes the temporary one.
5.  Cleans the structure (flattens 1‑request folders, sorts items, injects
    doc links & org headers).
6.  Merges the cleaned collection into an **existing** Postman collection of
    the same name, preserving manual edits.
7.  Optionally splits very large specs into sub‑collections (see `split_by_tag`)
    to avoid time‑outs.

Prereqs
───────
    pip install requests pyyaml
    export POSTMAN_API_KEY=***
    export POSTMAN_WORKSPACE_ID=***

This script is intentionally minimal: remove, extend, or swap out helpers to
match your organisation’s needs.
"""

from __future__ import annotations
import json
import os
import sys
import time
from pathlib import Path
from typing import Any, Dict, List
import requests
import yaml

# --------------------------------------------------------------------------- #
# Configuration – adapt to your org                                           #
# --------------------------------------------------------------------------- #

POSTMAN_API_KEY = os.getenv("POSTMAN_API_KEY")          # required
WORKSPACE_ID     = os.getenv("POSTMAN_WORKSPACE_ID")    # required
OPENAPI_PATH     = Path("openapi.json")                 # or .yaml
BASE_URL         = "https://api.getpostman.com"
BASE_DOC_URL     = "https://your.docs.host/#tag/"

# Headers you want every request to include
COMMON_HEADERS: Dict[str, Dict[str, Any]] = {
    "X‑Tenant": {"value": "{{tenantId}}", "disabled": False},
    "X‑App":    {"value": "{{appSlug}}",  "disabled": False},
}

# --------------------------------------------------------------------------- #
# 1. Load & sanitise OpenAPI                                                  #
# --------------------------------------------------------------------------- #

def load_openapi(path: Path) -> Dict[str, Any]:
    """Load JSON or YAML OpenAPI file into a Python dict."""
    if path.suffix in {".yaml", ".yml"}:
        return yaml.safe_load(path.read_text())
    return json.loads(path.read_text())

def replace_enums(spec: Dict[str, Any], placeholder: str = "<enum>") -> None:
    """Replace enum values with a placeholder to avoid churn in Postman diffs."""
    if isinstance(spec, dict):
        for k, v in spec.items():
            if k == "enum" and isinstance(v, list):
                spec[k] = [placeholder]
            else:
                replace_enums(v, placeholder)
    elif isinstance(spec, list):
        for item in spec:
            replace_enums(item, placeholder)

# --------------------------------------------------------------------------- #
# 2. Minimal Postman‑API helpers                                              #
# --------------------------------------------------------------------------- #

def _headers(json_body: bool = False) -> Dict[str, str]:
    hdr = {"X-API-Key": POSTMAN_API_KEY}
    if json_body:
        hdr["Content-Type"] = "application/json"
    return hdr

def import_openapi(spec: Dict[str, Any]) -> str:
    """Import spec → returns temporary collection UID."""
    url = f"{BASE_URL}/import/openapi?workspace={WORKSPACE_ID}"
    payload = {"type": "json", "input": spec, "options": {}}
    resp = requests.post(url, headers=_headers(True), json=payload)
    resp.raise_for_status()
    return resp.json()["collections"][0]["uid"]

def get_collection(col_id: str) -> Dict[str, Any]:
    resp = requests.get(f"{BASE_URL}/collections/{col_id}", headers=_headers())
    resp.raise_for_status()
    return resp.json()

def delete_collection(col_id: str) -> None:
    requests.delete(f"{BASE_URL}/collections/{col_id}", headers=_headers())

def update_collection(col_id: str, data: Dict[str, Any]) -> None:
    url = f"{BASE_URL}/collections/{col_id}"
    requests.put(url, headers=_headers(True), json=data).raise_for_status()

def get_workspace_collections() -> Dict[str, str]:
    """Return map {collection‑name: collection‑id} in the workspace."""
    ws = requests.get(f"{BASE_URL}/workspaces/{WORKSPACE_ID}", headers=_headers()).json()
    return {c["name"]: c["id"] for c in ws["workspace"].get("collections", [])}

# --------------------------------------------------------------------------- #
# 3. Collection transformers (flatten, sort, headers, docs)                   #
# --------------------------------------------------------------------------- #

def flatten_single_folders(item: Dict[str, Any]) -> None:
    if "item" not in item:
        return
    for child in list(item["item"]):
        flatten_single_folders(child)
    if len(item["item"]) == 1:
        child = item["item"][0]
        if "request" in child and child["name"] == item["name"]:
            item.clear(); item.update(child)

def sort_items_alpha(item: Dict[str, Any]) -> None:
    if "item" in item:
        item["item"].sort(key=lambda i: i["name"])
        for ch in item["item"]:
            sort_items_alpha(ch)

def apply_common_headers(element: Any) -> None:
    if isinstance(element, dict):
        if "header" in element:
            for hdr in element["header"]:
                if hdr["key"] in COMMON_HEADERS:
                    hdr.update(COMMON_HEADERS[hdr["key"]])
        for v in element.values():
            apply_common_headers(v)
    elif isinstance(element, list):
        for itm in element:
            apply_common_headers(itm)

def add_doc_links(item: Dict[str, Any]) -> None:
    if "item" in item:
        for ch in item["item"]:
            add_doc_links(ch)
    elif "request" in item and "url" in item["request"]:
        path = item["request"]["url"].get("path", [])
        if path:
            link = f"{BASE_DOC_URL}{'.'.join(path[:-1])}/operation/{path[-1]}"
            item["request"]["description"] = (item["request"].get("description","") + 
                                              f"\n[Docs]({link})").lstrip()

# --------------------------------------------------------------------------- #
# 4. Merge (name‑based, ID‑agnostic)                                          #
# --------------------------------------------------------------------------- #

def merge_items(old: List[Dict], new: List[Dict]) -> List[Dict]:
    old_map = {i["name"]: i for i in old}
    result: List[Dict] = []

    for n in new:
        name = n["name"]
        if name in old_map:
            o = old_map[name]
            if "item" in n:
                o["item"] = merge_items(o.get("item", []), n["item"])
            o.update({k: v for k, v in n.items() if k not in {"id", "uid", "item"}})
            result.append(o)
        else:
            n.pop("id", None); n.pop("uid", None)
            result.append(n)

    return result

# --------------------------------------------------------------------------- #
# 5. Pipeline                                                                 #
# --------------------------------------------------------------------------- #

def main() -> None:
    if not POSTMAN_API_KEY or not WORKSPACE_ID:
        sys.exit("POSTMAN_API_KEY and POSTMAN_WORKSPACE_ID env vars are required.")

    spec = load_openapi(OPENAPI_PATH)
    replace_enums(spec)

    tmp_id  = import_openapi(spec)
    new_col = get_collection(tmp_id)
    delete_collection(tmp_id)            # keep workspace clean

    root = new_col["collection"]
    flatten_single_folders(root)
    sort_items_alpha(root)
    apply_common_headers(root)
    add_doc_links(root)

    existing_map = get_workspace_collections()
    col_name     = root["info"]["name"]

    if col_name in existing_map:
        old_col = get_collection(existing_map[col_name])
        old_col["collection"]["item"] = merge_items(
            old_col["collection"]["item"], root["item"]
        )
        update_collection(existing_map[col_name], old_col)
        print(f"✅ Updated collection “{col_name}”.")
    else:
        # wipe Postman‑generated IDs for portability
        for itm in root["item"]:
            itm.pop("id", None); itm.pop("uid", None)
        requests.post(
            f"{BASE_URL}/collections?workspace={WORKSPACE_ID}",
            headers=_headers(True),
            json={"collection": root},
        ).raise_for_status()
        print(f"✅ Created new collection “{col_name}”.")
    print("Done.")

if __name__ == "__main__":
    start = time.time()
    main()
    print(f"Completed in {time.time() - start:.1f}s")