Kyle
f32203b630
- FastAPI + SQLite backend with routers for diagrams, devices, wires, bundles, connectors, export, and Octopart/Nexar search - Konva.js canvas: ortho/direct/curved routing, wire crossings, harness mode, snap-to-grid, multi-select, drag, resize, undo - Wire features: color/stripe, twisted pair helix, shielded glow, gauge size label, multi-core bundle grouping, length/unit tracking - Device library: connector, terminal block, cable, splice, label, group, relay, fuse, switch, component with custom connector support - Exports: BOM CSV, assembly TXT, JSON, formboard layout, PNG image - Auto-migration for schema changes via ALTER TABLE at startup Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
import os
|
|
import time
|
|
from typing import Tuple
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
|
|
router = APIRouter(tags=["octopart"])
|
|
|
|
NEXAR_TOKEN_URL = "https://identity.nexar.com/connect/token"
|
|
NEXAR_API_URL = "https://api.nexar.com/graphql"
|
|
|
|
_token_cache: dict = {"token": None, "expires_at": 0.0}
|
|
|
|
_SEARCH_GQL = """
|
|
query PartSearch($q: String!, $limit: Int!) {
|
|
supSearchMpn(q: $q, limit: $limit) {
|
|
results {
|
|
part {
|
|
mpn
|
|
manufacturer { name }
|
|
shortDescription
|
|
bestDatasheet { url }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def _credentials() -> Tuple[str, str]:
|
|
return (
|
|
os.getenv("NEXAR_CLIENT_ID", ""),
|
|
os.getenv("NEXAR_CLIENT_SECRET", ""),
|
|
)
|
|
|
|
|
|
async def _get_token() -> str:
|
|
client_id, client_secret = _credentials()
|
|
if not client_id or not client_secret:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Nexar/Octopart credentials not configured — set NEXAR_CLIENT_ID and NEXAR_CLIENT_SECRET environment variables",
|
|
)
|
|
|
|
now = time.time()
|
|
if _token_cache["token"] and now < _token_cache["expires_at"] - 60:
|
|
return _token_cache["token"] # type: ignore[return-value]
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
NEXAR_TOKEN_URL,
|
|
data={
|
|
"grant_type": "client_credentials",
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
},
|
|
timeout=10,
|
|
)
|
|
if resp.status_code != 200:
|
|
raise HTTPException(status_code=502, detail=f"Nexar auth failed: {resp.text[:200]}")
|
|
|
|
body = resp.json()
|
|
_token_cache["token"] = body["access_token"]
|
|
_token_cache["expires_at"] = now + body.get("expires_in", 3600)
|
|
return _token_cache["token"] # type: ignore[return-value]
|
|
|
|
|
|
@router.get("/status")
|
|
async def octopart_status():
|
|
client_id, client_secret = _credentials()
|
|
return {"configured": bool(client_id and client_secret)}
|
|
|
|
|
|
@router.get("/search")
|
|
async def octopart_search(
|
|
q: str = Query(..., min_length=1),
|
|
limit: int = Query(8, ge=1, le=20),
|
|
):
|
|
token = await _get_token()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
NEXAR_API_URL,
|
|
headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
|
|
json={"query": _SEARCH_GQL, "variables": {"q": q, "limit": limit}},
|
|
timeout=12,
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
raise HTTPException(status_code=502, detail=f"Nexar API error {resp.status_code}")
|
|
|
|
body = resp.json()
|
|
if "errors" in body:
|
|
raise HTTPException(status_code=502, detail=str(body["errors"])[:300])
|
|
|
|
results = body.get("data", {}).get("supSearchMpn", {}).get("results", [])
|
|
out = []
|
|
for r in results:
|
|
part = r.get("part")
|
|
if not part:
|
|
continue
|
|
out.append({
|
|
"mpn": part.get("mpn", ""),
|
|
"manufacturer": (part.get("manufacturer") or {}).get("name", ""),
|
|
"description": part.get("shortDescription") or "",
|
|
"datasheet_url": (part.get("bestDatasheet") or {}).get("url"),
|
|
})
|
|
return out
|