Making Requests¶

netbox_sdk supports two async request styles:

the low-level NetBoxApiClient for direct HTTP calls
the higher-level facade returned by api() for PyNetBox-style workflows
the versioned typed client returned by typed_api() for validated Pydantic I/O

Both layers are async and can be used in the same project.

Facade usage¶

The facade is the shortest path for common NetBox SDK operations:

from netbox_sdk import api

nb = api("https://netbox.example.com", token="mytoken")

Retrieve a single record¶

device = await nb.dcim.devices.get(42)
if device is not None:
    print(device.name)

Filter and iterate¶

devices = nb.dcim.devices.filter(site="lon", role="leaf-switch")

async for device in devices:
    print(device.name)

Create, modify, and save¶

device = await nb.dcim.devices.create(
    name="sw-lon-01",
    site={"id": 1},
    role={"id": 2},
    device_type={"id": 3},
)

device.name = "sw-lon-01-renamed"
await device.save()

Use detail endpoints¶

prefix = await nb.ipam.prefixes.get(123)

available = await prefix.available_ips.list()
new_ip = await prefix.available_ips.create({})
new_prefix = await prefix.available_prefixes.create({"prefix_length": 28})

Rack elevation JSON vs SVG¶

rack = await nb.dcim.racks.get(5)
units = await rack.elevation.list()
svg = await rack.elevation.list(render="svg")

Plugins¶

plugins = await nb.plugins.installed_plugins()

async for olt in nb.plugins.gpon.olts.filter(status="active"):
    print(olt.name)

Typed client usage¶

Use the typed client when you want validated request payloads, validated response payloads, and version-specific endpoint models.

from netbox_sdk import typed_api

nb = typed_api(
    "https://netbox.example.com",
    token="mytoken",
    netbox_version="4.5",
)

Retrieve a typed record¶

device = await nb.dcim.devices.get(42)
if device is not None:
    print(device.name)

Validate a request before HTTP¶

from netbox_sdk import TypedRequestValidationError

try:
    await nb.ipam.prefixes.available_ips.create(7, body=[{"prefix_length": "invalid"}])
except TypedRequestValidationError as exc:
    print(exc)

Version selection¶

typed_api("https://netbox.example.com", token="tok", netbox_version="4.5")
typed_api("https://netbox.example.com", token="tok", netbox_version="4.5.5")

The typed client supports NetBox 4.5, 4.4, and 4.3. Generated models for those release lines are committed in the repository and shipped with the package.

Low-level client usage¶

All direct HTTP calls go through NetBoxApiClient.request().

Basic usage¶

from netbox_sdk import Config, NetBoxApiClient

cfg = Config(base_url="https://netbox.example.com", token_version="v1", token_secret="mytoken")
client = NetBoxApiClient(cfg)

GET — list resources¶

response = await client.request("GET", "/api/dcim/devices/")
data = response.json()

print(data["count"])    # total number of results
print(data["results"])  # list of device objects

GET — filter results¶

response = await client.request(
    "GET",
    "/api/dcim/devices/",
    query={"site": "lon", "role": "leaf-switch", "limit": "50"},
)

GET — retrieve single object¶

response = await client.request("GET", "/api/dcim/devices/42/")
device = response.json()
print(device["name"])

POST — create¶

response = await client.request(
    "POST",
    "/api/dcim/devices/",
    payload={
        "name": "sw-lon-01",
        "device_type": {"id": 3},
        "site": {"id": 1},
        "role": {"id": 2},
    },
)
if response.status == 201:
    new_device = response.json()
    print(f"Created device ID {new_device['id']}")

PUT — full update¶

response = await client.request(
    "PUT",
    "/api/dcim/devices/42/",
    payload={
        "name": "sw-lon-01-renamed",
        "device_type": {"id": 3},
        "site": {"id": 1},
        "role": {"id": 2},
    },
)

PATCH — partial update¶

response = await client.request(
    "PATCH",
    "/api/dcim/devices/42/",
    payload={"name": "sw-lon-01-renamed"},
)

DELETE¶

response = await client.request("DELETE", "/api/dcim/devices/42/")
if response.status == 204:
    print("Deleted")

Response object¶

request() always returns ApiResponse:

class ApiResponse(BaseModel):
    status: int               # HTTP status code
    text: str                 # raw response body
    headers: dict[str, str]   # response headers

    def json(self) -> Any: ...  # parse text as JSON

Check the status code before parsing:

response = await client.request("GET", "/api/dcim/devices/99/")
if response.status == 404:
    print("Device not found")
elif response.status == 200:
    device = response.json()

Pagination¶

NetBox list endpoints return paginated results. Iterate pages manually:

async def list_all_devices(client: NetBoxApiClient) -> list[dict]:
    results = []
    offset = 0
    limit = 100
    while True:
        response = await client.request(
            "GET",
            "/api/dcim/devices/",
            query={"limit": str(limit), "offset": str(offset)},
        )
        data = response.json()
        results.extend(data["results"])
        if not data.get("next"):
            break
        offset += limit
    return results

GraphQL¶

query = """
{
  device_list(site: "lon") {
    id
    name
    device_type { model }
    primary_ip4 { address }
  }
}
"""
response = await client.graphql(query)
data = response.json()["data"]["device_list"]

With variables:

query = "query($id: Int!) { device(id: $id) { name status } }"
response = await client.graphql(query, variables={"id": 42})

Connection health check¶

Before making API calls, verify connectivity:

from netbox_sdk import ConnectionProbe

probe = await client.probe_connection()

if probe.ok:
    print(f"Connected — NetBox API {probe.version}")
else:
    print(f"Failed (HTTP {probe.status}): {probe.error}")

probe_connection() returns ok=True for any reachable status below 400, or 403 (which means the URL is valid but the token is wrong — still reachable).

HTTP caching¶

GET requests to /api/... paths are automatically cached to disk under the NetBox SDK config root, typically ~/.config/netbox-sdk/http-cache/. The X-NBX-Cache response header shows the cache outcome:

Value	Meaning
`MISS`	First fetch, stored to cache
`HIT`	Served from fresh cache
`REVALIDATED`	Server returned 304, cache TTL extended
`STALE`	Network error; stale data served

Tune the cache policy:

from netbox_sdk.http_cache import CachePolicy, HttpCacheStore
from netbox_sdk.config import cache_dir

# Default: 60s fresh, 300s stale-if-error for list endpoints
store = HttpCacheStore(cache_dir())

POST/PUT/PATCH/DELETE requests are never cached.

Schema-driven requests¶

Use netbox_sdk.services to resolve action names to HTTP calls:

from netbox_sdk import build_schema_index
from netbox_sdk.services import resolve_dynamic_request, run_dynamic_command

idx = build_schema_index()

# Resolve to a ResolvedRequest
req = resolve_dynamic_request(
    idx, "dcim", "devices", "list",
    object_id=None, query={"site": "lon"}, payload=None,
)
# req.method == "GET", req.path == "/api/dcim/devices/", req.query == {"site": "lon"}

# Or run it directly
response = await run_dynamic_command(
    client, idx, "dcim", "devices", "list",
    object_id=None,
    query_pairs=["site=lon"],
    body_json=None,
    body_file=None,
)

Supported actions: list, get, create, update, patch, delete.

Multipart uploads¶

NetBoxApiClient.request() automatically switches to multipart form data when a payload contains file-like values:

with open("rack-photo.png", "rb") as handle:
    response = await client.request(
        "POST",
        "/api/extras/image-attachments/",
        payload={
            "object_type": "dcim.device",
            "object_id": 42,
            "name": "Front view",
            "image": ("rack-photo.png", handle, "image/png"),
        },
    )

Supported file inputs include plain file objects and tuples in the form (filename, file_obj) or (filename, file_obj, content_type).

Status and OpenAPI helpers¶

The low-level client now exposes convenience helpers for common API metadata:

status = await client.status()
version = await client.get_version()
spec = await client.openapi()

Token provisioning is also available:

token_response = await client.create_token("admin", "password")
print(token_response.json()["key"])