dstephenson
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
# nocodb_client.py - Fetch switch list from NocoDB "Fiber Switches" base
|
|
import logging
|
|
import requests
|
|
from config import NOCODB_URL, NOCODB_TOKEN
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BASE_ID = "p1h7iyzjjbsnfv5"
|
|
TABLE_ID = "md03qoibd5fz839"
|
|
|
|
HEADERS = {
|
|
"xc-token": NOCODB_TOKEN,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def get_switches(dept: str = None) -> list[dict]:
|
|
"""
|
|
Return list of switch dicts from NocoDB (Active=YES only).
|
|
dept: None = all, "ELEC" or "GW" to filter by department.
|
|
Raises RuntimeError if NocoDB is unreachable or returns no rows.
|
|
"""
|
|
where = "(Active,eq,YES)"
|
|
if dept:
|
|
where = f"(Active,eq,YES)~and(Dept,eq,{dept})"
|
|
|
|
switches = []
|
|
offset = 0
|
|
limit = 100
|
|
|
|
while True:
|
|
r = requests.get(
|
|
f"{NOCODB_URL}/api/v1/db/data/noco/{BASE_ID}/{TABLE_ID}",
|
|
headers=HEADERS,
|
|
params={"limit": limit, "offset": offset, "where": where},
|
|
timeout=15,
|
|
)
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"NocoDB error {r.status_code}: {r.text[:200]}")
|
|
|
|
data = r.json()
|
|
for row in data.get("list", []):
|
|
ip = (row.get("IP") or "").strip()
|
|
if not ip:
|
|
continue
|
|
switches.append({
|
|
"ip": ip,
|
|
"hostname": (row.get("Hostname") or "").strip(),
|
|
"location": (row.get("Location") or "").strip(),
|
|
"model": (row.get("Model") or "").strip(),
|
|
"manufacturer": (row.get("Manufacturer") or "FS").strip(),
|
|
"dept": (row.get("Dept") or "").strip(),
|
|
"asset_tag": (row.get("Asset Tag") or "").strip(),
|
|
})
|
|
|
|
if data.get("pageInfo", {}).get("isLastPage", True):
|
|
break
|
|
offset += limit
|
|
|
|
if not switches:
|
|
raise RuntimeError(f"NocoDB returned 0 active switches (dept={dept})")
|
|
|
|
logger.info(f"NocoDB: loaded {len(switches)} switches" + (f" (dept={dept})" if dept else ""))
|
|
return switches
|
|
|
|
|
|
def get_switch_ips(dept: str = None) -> list[str]:
|
|
"""Convenience wrapper — returns just the IP list."""
|
|
return [s["ip"] for s in get_switches(dept=dept)]
|