Skip to content

common-patterns.md

Latest commit

 

History

History
406 lines (323 loc) · 14 KB

common-patterns.md

File metadata and controls

406 lines (323 loc) · 14 KB
Raw

Common Code Patterns

Authentication Setup

Always validate env vars first

from utils.env_check import check_env, get_api_keys_from_env

check_env()  # Exits with OS-specific instructions if vars missing
api_keys = get_api_keys_from_env()  # Returns SDK-compatible dict

Using the NCM SDK

from ncm import ncm
from utils.env_check import check_env, get_api_keys_from_env

check_env()
client = ncm.NcmClient(api_keys=get_api_keys_from_env())

Using the Session Utility (for direct API calls)

import os
from utils.env_check import check_env
from utils.session import APISession
from utils.logger import get_logger

check_env()
logger = get_logger('my_script')
session = APISession(
    logger=logger,
    cp_api_id=os.environ['X_CP_API_ID'],
    cp_api_key=os.environ['X_CP_API_KEY'],
    ecm_api_id=os.environ['X_ECM_API_ID'],
    ecm_api_key=os.environ['X_ECM_API_KEY'],
)

Pagination

With NCM SDK (automatic)

# SDK methods handle pagination internally
routers = client.get_routers()  # returns all routers

With requests (manual)

import requests
from utils.env_check import check_env, get_api_keys_from_env

check_env()
api_keys = get_api_keys_from_env()

base_url = 'https://www.cradlepointecm.com/api/v2'
headers = {k: v for k, v in api_keys.items() if k != 'token'}
headers['Content-Type'] = 'application/json'

def get_all(endpoint, params=None):
    url = f'{base_url}/{endpoint}/'
    results = []
    while url:
        resp = requests.get(url, headers=headers, params=params)
        resp.raise_for_status()
        data = resp.json()
        results.extend(data.get('data', []))
        url = data.get('meta', {}).get('next')
        params = None  # params already in next URL
    return results

With Session Utility (automatic via generator)

with APISession(logger=logger, **creds) as session:
    for router in session.get('routers'):
        process(router)

Pagination (v3 — cursor-based)

API v3 uses cursor-based pagination, not offset-based like v2. The max page size is 50. Follow links.next until it's absent.

import httpx

def get_all_v3(path, headers, params=None):
    """Fetch all pages from a v3 cursor-paginated endpoint."""
    base = "https://api.cradlepointecm.com/api/v3"
    params = params or {}
    params.setdefault("page[size]", 50)
    results = []
    url = f"{base}{path}"

    while url:
        resp = httpx.get(url, headers=headers, params=params)
        resp.raise_for_status()
        body = resp.json()
        for item in body.get("data", []):
            record = {"id": item["id"], **item.get("attributes", {})}
            results.append(record)
        url = body.get("links", {}).get("next")
        params = None  # params are baked into the cursor URL
    return results

# Usage
headers = {
    "Authorization": "Bearer <token>",
    "Accept": "application/vnd.api+json",
}
assets = get_all_v3("/asset_endpoints", headers)
subscriptions = get_all_v3("/subscriptions", headers)

Error Handling

import requests
from time import sleep

def api_call_with_retry(func, max_retries=5, backoff=2):
    for attempt in range(max_retries):
        try:
            return func()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code in (408, 409, 429, 500, 502, 503, 504):
                wait = backoff ** attempt
                if e.response.status_code == 429:
                    wait = float(e.response.headers.get("Retry-After", wait))
                sleep(wait)
                continue
            raise
    raise Exception(f"Failed after {max_retries} retries")

CSV Export Pattern

import csv

def export_to_csv(data, filename, fields):
    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fields, extrasaction='ignore')
        writer.writeheader()
        writer.writerows(data)

Filtering Routers

# By state
online_routers = client.get_routers(state='online')

# By group
group_routers = client.get_routers_for_group(group_id=123)

# By account
account_routers = client.get_routers_for_account(account_id=456)

# Specific fields only
routers = client.get_routers(fields='id,name,state,mac')

Configuration Push Pattern

def push_config_to_routers(client, router_ids, config):
    """Push a configuration to multiple routers."""
    results = []
    for router_id in router_ids:
        try:
            result = client.patch_configuration_managers(router_id, config)
            results.append({'router_id': router_id, 'status': 'success'})
        except Exception as e:
            results.append({'router_id': router_id, 'status': 'error', 'error': str(e)})
    return results

Copying a Config Subtree Between Groups

Copying one branch of a group config (a MAC filter, an identity set, WAN rules) from a "master" group to others. Two asymmetries make this trickier than it looks:

  1. On read, NCM returns config arrays as index-keyed objects — {"0": {...}, "1": {...}}, not [{...}, {...}]. Normalize before reasoning about the list.
  2. On write, PATCH merges objects but replaces arrays entirely. That is the lever for choosing mirror-vs-merge semantics: send the same data as a JSON array to replace the destination list outright, or as an index-keyed object to overwrite position-by-position and leave extra destination entries in place.
def extract_subtree(configuration, *path):
    """Pull a branch out of a group's [updates, removals] config diff."""
    if not isinstance(configuration, list) or not configuration:
        return None
    node = configuration[0]
    for key in path:
        if not isinstance(node, dict):
            return None
        node = node.get(key)
    return node if isinstance(node, dict) else None


def normalize_entries(value):
    """Index-keyed object OR real array -> ordered list of dicts."""
    if isinstance(value, dict):
        keys = sorted(value, key=lambda k: (0, int(k)) if str(k).isdigit() else (1, k))
        return [dict(value[k]) for k in keys if isinstance(value[k], dict)]
    if isinstance(value, list):
        return [dict(v) for v in value if isinstance(v, dict)]
    return []


# Read the source branch
src = client.get_groups(id=master_id, fields='id,name,configuration')[0]
macfilter = extract_subtree(src['configuration'], 'firewall', 'macfilter')
entries = normalize_entries(macfilter.get('macs'))

# mirror: array -> destination list is replaced wholesale
# merge:  index-keyed object -> per-index overwrite, extras survive
macs = entries if mirror else {str(i): e for i, e in enumerate(entries)}

payload = {'configuration': [{'firewall': {'macfilter': {
    'enabled': macfilter.get('enabled'),
    'whitelist': macfilter.get('whitelist'),
    'macs': macs,
}}}, []]}

for gid in destination_ids:
    try:
        client.patch_group_configuration(gid, payload)
    except Exception as e:      # keep going; one bad group must not abort the batch
        log_failure(gid, e)

Send only the branch you intend to copy. Building the payload from the extracted subtree (rather than forwarding the whole configuration[0]) keeps unrelated source settings from riding along into the destinations.

UUID-keyed collections copy differently than index-keyed ones

The example above is an index-keyed array. Collections that support _id_ (identities.ip/mac/port, lan, vpn.tunnels, security.zfw.zones, wan.rules, and the rest of the list in api-configuration.md) are keyed by UUID instead, and that changes what a copy means:

  • Index-keyed (macs, members): positions are meaningful, so sending an array replaces the list.
  • UUID-keyed (identities.ip): PATCH merges by key, so the updates dict alone only adds the source entries and overwrites any sharing a UUID. Entries existing only in the destination have no matching key and would survive.

An updates-only PATCH is therefore additive. To make a destination equal the source, add a removals list (second diff element) — PATCH honors it, so an exact mirror is achievable without PUT. See "Mirroring a collection" below.

Real subtrees nest the two styles, so one copy touches both levels. In identities.ip the outer collection is UUID-keyed while each entry's members is an index-keyed array — the outer level is always a merge, and the mirror/merge choice applies to the inner address list:

identities = extract_subtree(src['configuration'], 'identities')['ip']

out = {}
for key, entry in identities.items():
    identity_id = entry.get('_id_') or key      # _id_ wins if they disagree
    copied = dict(entry)
    copied['_id_'] = identity_id                # required inside the object too
    members = normalize_entries(entry.get('members'))
    copied['members'] = members if mirror else {str(i): m for i, m in enumerate(members)}
    out[identity_id] = copied                   # key must equal _id_

payload = {'configuration': [{'identities': {'ip': out}}, []]}

Key the output dict by _id_ rather than by whatever key you read it under. The two normally agree, but if they ever diverge, NCM validates against the _id_ inside the object.

Mirroring a collection (not just copying it)

To make the destination equal the source, read the destination too and emit removals for whatever it has that the source does not. This is the same shape NCM's own UI sends. Remember that removal paths address array positions with integer indices, while the updates dict uses string keys for the same positions:

def build_mirror_payload(src_identities, dst_identities):
    """-> (payload, plan). Mirrors identities.ip onto one destination group."""
    src = {e.get('_id_') or k: e for k, e in (src_identities or {}).items()}
    dst = {e.get('_id_') or k: e for k, e in (dst_identities or {}).items()}

    updates, removals = {}, []

    for identity_id, entry in src.items():
        members = normalize_entries(entry.get('members'))
        copied = {k: v for k, v in entry.items() if k != 'members'}
        copied['_id_'] = identity_id
        copied['members'] = {str(i): m for i, m in enumerate(members)}   # string keys
        updates[identity_id] = copied

        dst_entry = dst.get(identity_id)
        if dst_entry:
            dst_members = normalize_entries(dst_entry.get('members'))
            # Drop surplus positions, highest index first
            for index in range(len(dst_members) - 1, len(members) - 1, -1):
                removals.append(['identities', 'ip', identity_id, 'members', index])

    # Drop destination-only entries wholesale
    for identity_id in dst:
        if identity_id not in src:
            removals.append(['identities', 'ip', identity_id])

    return {'configuration': [{'identities': {'ip': updates}}, removals]}

This costs one extra GET per destination, since removals depend on each destination's current contents — the payload is no longer identical across groups. Skip the PATCH entirely when a destination already matches, and compute a per-group summary of what changed so a dry-run mode can show it before anything is sent.

Prefer PATCH over PUT here even when removing things: PATCH honors the removals list, whereas PUT additionally resets every unmentioned field to defaults, which at /groups/{id}/ scope means wiping unrelated group settings. If you do need a real PUT, note that put_group_configuration() raises AttributeError after the write lands; see the entry in known-issues.md.

Date Filtering Pattern

from datetime import datetime, timedelta

# Get alerts from last 24 hours
yesterday = (datetime.utcnow() - timedelta(hours=24)).strftime('%Y-%m-%dT%H:%M:%S')
alerts = client.get_router_alerts(created_at__gt=yesterday)

Batch Operations

def batch_operation(items, batch_size=50, operation=None):
    """Process items in batches."""
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        for item in batch:
            operation(item)

Web UI Template

When building any web interface in this project, use the web_app_template located at web_apps/web_app_template/ as the style foundation. It provides a complete, consistent design system including layout, components, and theming.

Reference files:

  • web_apps/web_app_template/index.html — HTML structure
  • web_apps/script_manager/static/css/style.css — Full CSS with light/dark mode
  • web_apps/script_manager/static/js/app.js — JS patterns (dark mode toggle, sidebar, etc.)

All web apps must support light mode and dark mode:

  • Use CSS custom properties (var(--*)) for all colors
  • Toggle via body.dark-mode class
  • Persist preference in localStorage
  • Include both logo.png and logo_dark.png with automatic swap

See .kiro/steering/web-ui-standards.md for the full checklist and CSS variable reference.

NCM SDK with FastAPI (async) — Avoiding Event Loop Blocking

The NCM SDK uses synchronous requests.Session internally. Calling SDK methods directly from async def FastAPI endpoints blocks the entire event loop, making the server unresponsive to all requests (including health checks, static files, and Ctrl+C) for the duration of the API call (often 10–30 seconds for large accounts).

Always wrap SDK calls in run_in_executor:

import asyncio
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

def _fetch_data():
    """Synchronous function that calls the NCM SDK."""
    client = ncm.NcmClient(api_keys=api_keys)
    return client.get_routers()

@app.get("/api/data")
async def get_data():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, _fetch_data)
    return JSONResponse({"data": result})

This runs the blocking SDK call in a thread pool, keeping the event loop free to serve other requests, handle WebSocket connections, and respond to shutdown signals. Apply this pattern to ALL endpoints that call _get_cellular_health() or any other function using the NCM SDK.

Also applies to: SQLite writes, file I/O on large files, or any other blocking operation inside an async handler.