Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co

## Project

`aiopynetbox` - a fully async NetBox API client, built from scratch with httpx. It is inspired by [pynetbox](https://github.com/netbox-community/pynetbox) (the popular sync client) but is **not a port**: pynetbox's core ergonomics depend on sync-only Python protocols that cannot be awaited, so the API surface here is deliberately different (see Design constraints below).
`aiopynetbox` - a fully async NetBox API client, built from scratch with httpx2 (the Pydantic-maintained fork of httpx). It is inspired by [pynetbox](https://github.com/netbox-community/pynetbox) (the popular sync client) but is **not a port**: pynetbox's core ergonomics depend on sync-only Python protocols that cannot be awaited, so the API surface here is deliberately different (see Design constraints below).

Package layout: `src/aiopynetbox/`, tests in `tests/`. Managed with `uv`.

Expand All @@ -31,7 +31,7 @@ pynetbox ideas worth keeping (they're pure Python, no I/O): app/endpoint attribu

## Architecture

All HTTP funnels through `Api._request_response()` ([api.py](src/aiopynetbox/api.py)) - auth headers (v1 `Token`/v2 `nbt_` `Bearer`) and error raising (POST 409 -> `AllocationError`, everything else non-success -> `RequestError`) live there and nowhere else, as does the retry loop (429 retried for any method honoring Retry-After; 502/503/504 and httpx.TransportError retried for GET only, since ambiguous writes may have been processed; `Api(retries=)` bounds attempts, `_backoff()` does capped exponential backoff with jitter); `Api._request()` adds JSON decoding (`_decode` -> `ContentError`). Detail-path callers (`Endpoint.get(id)`, `full_details()`, `save()`) use `_request_response` directly to capture the `ETag` header: records store it as `_etag` and `save()` sends `If-Match` (NetBox 4.6+ optimistic locking; stale ETag -> 412 `RequestError`); repeat `full_details()` calls send `If-None-Match` and a 304 (allowed through `_request_response` only when that header was sent) keeps current data without re-parsing). `App.__getattr__` ([app.py](src/aiopynetbox/app.py)) turns any attribute into an `Endpoint` ([endpoint.py](src/aiopynetbox/endpoint.py)), which builds URLs (`_`->`-`) and returns `Record`/`RecordSet` ([response.py](src/aiopynetbox/response.py)). `PluginsApp` (also app.py) routes `nb.plugins.<plugin>` into `/api/plugins/<plugin>/`.
All HTTP funnels through `Api._request_response()` ([api.py](src/aiopynetbox/api.py)) - auth headers (v1 `Token`/v2 `nbt_` `Bearer`) and error raising (POST 409 -> `AllocationError`, everything else non-success -> `RequestError`) live there and nowhere else, as does the retry loop (429 retried for any method honoring Retry-After; 502/503/504 and httpx2.TransportError retried for GET only, since ambiguous writes may have been processed; `Api(retries=)` bounds attempts, `_backoff()` does capped exponential backoff with jitter); `Api._request()` adds JSON decoding (`_decode` -> `ContentError`). Detail-path callers (`Endpoint.get(id)`, `full_details()`, `save()`) use `_request_response` directly to capture the `ETag` header: records store it as `_etag` and `save()` sends `If-Match` (NetBox 4.6+ optimistic locking; stale ETag -> 412 `RequestError`); repeat `full_details()` calls send `If-None-Match` and a 304 (allowed through `_request_response` only when that header was sent) keeps current data without re-parsing). `App.__getattr__` ([app.py](src/aiopynetbox/app.py)) turns any attribute into an `Endpoint` ([endpoint.py](src/aiopynetbox/endpoint.py)), which builds URLs (`_`->`-`) and returns `Record`/`RecordSet` ([response.py](src/aiopynetbox/response.py)). `PluginsApp` (also app.py) routes `nb.plugins.<plugin>` into `/api/plugins/<plugin>/`.

`Endpoint.__init__` resolves its Record subclass from `ENDPOINT_MODELS` in [models.py](src/aiopynetbox/models.py) (`"<app>/<endpoint>"` keys, e.g. `ipam/prefixes` -> `Prefixes` with `available_ips`/`available_prefixes` properties returning a `DetailEndpoint`; `core/data-sources` -> `DataSources` with a `sync` trigger). `register_model()` is the public way to add entries (plugin endpoints use the `"plugins/<plugin>"` app key). `DetailEndpoint.list()` reuses `RecordSet` - its plain-list branch handles non-paginated detail routes. `App.endpoint(name)` bypasses the `_` -> `-` slug conversion for literal-underscore endpoints. Import order matters: api -> app -> endpoint -> models -> response; response only TYPE_CHECKING-imports the others.

Expand All @@ -49,12 +49,12 @@ Key mechanics in `response.py`:

The package is fully type-annotated (`from __future__ import annotations` everywhere) and ships `py.typed`. `__version__` comes from package metadata via `importlib.metadata`.

Tests run entirely against `FakeNetbox` in [tests/conftest.py](tests/conftest.py) - an in-memory NetBox behind `httpx.MockTransport` (no network, no mocking library). Extend it when adding endpoints/behaviors.
Tests run entirely against `FakeNetbox` in [tests/conftest.py](tests/conftest.py) - an in-memory NetBox behind `httpx2.MockTransport` (no network, no mocking library). Extend it when adding endpoints/behaviors.

Not implemented yet (deliberately, add only when needed): napalm helpers (NetBox dropped built-in napalm in 3.5), cable trace helpers, file uploads (multipart), OpenAPI filter validation.

## Conventions

- httpx `AsyncClient` is the only HTTP transport; the client should be usable as an async context manager (`async with aiopynetbox.api(...) as nb:`) so the connection pool is closed deterministically. The context manager is one-shot; `aclose()` closes only clients the Api created - a `client=` passed in is the caller's to close (httpx convention).
- httpx2 `AsyncClient` is the only HTTP transport; the client should be usable as an async context manager (`async with aiopynetbox.api(...) as nb:`) so the connection pool is closed deterministically. The context manager is one-shot; `aclose()` closes only clients the Api created - a `client=` passed in is the caller's to close (httpx2 convention).
- No sync wrapper/facade unless explicitly requested.
- A local reference clone of pynetbox may exist in the session scratchpad, not in this repo - never vendor pynetbox code without noting its Apache 2.0 license.
11 changes: 11 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,16 @@ All notable changes to this project are documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [0.2.0] - 2026-08-17

### Changed

- Switched the HTTP transport from httpx to
[httpx2](https://github.com/pydantic/httpx2), the Pydantic-maintained
fork (httpx development has stalled). The API is unchanged, but a
custom client passed via `client=` must now be an `httpx2.AsyncClient`
(`pip install httpx2`, `import httpx2`).

## [0.1.0] - 2026-07-24

Initial release.
Expand Down Expand Up @@ -57,4 +67,5 @@ Initial release.
- A runnable FastAPI example (`examples/fastapi_app.py`) showing the
app-state / lifespan usage pattern.

[0.2.0]: https://github.com/challey74/aiopynetbox/releases/tag/v0.2.0
[0.1.0]: https://github.com/challey74/aiopynetbox/releases/tag/v0.1.0
2 changes: 1 addition & 1 deletion CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ All four checks run in CI and must pass.
## Testing conventions

Tests run entirely against `FakeNetbox` in `tests/conftest.py`, an
in-memory NetBox served through `httpx.MockTransport`. Tests never touch
in-memory NetBox served through `httpx2.MockTransport`. Tests never touch
the network and never require a real NetBox instance. If your change needs
an endpoint or behavior the fake doesn't model yet, extend the fake.

Expand Down
12 changes: 6 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
[![Python versions](https://img.shields.io/pypi/pyversions/aiopynetbox)](https://pypi.org/project/aiopynetbox/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue)](LICENSE)

Fully async NetBox API client for Python, built on [httpx](https://www.python-httpx.org/).
Fully async NetBox API client for Python, built on [httpx2](https://github.com/pydantic/httpx2).

Inspired by [pynetbox](https://github.com/netbox-community/pynetbox), redesigned
for asyncio. This is not a port: pynetbox's core ergonomics (lazy attribute
Expand Down Expand Up @@ -70,7 +70,7 @@ that isn't loaded raises `AttributeError` telling you to

## Features

- **Explicit async everywhere**: `httpx.AsyncClient` under the hood, used as
- **Explicit async everywhere**: `httpx2.AsyncClient` under the hood, used as
an async context manager so the connection pool closes deterministically.
- **Concurrent pagination**: after the first page reveals the count, the
remaining pages are fetched in parallel (bounded by `max_concurrency`,
Expand Down Expand Up @@ -162,18 +162,18 @@ async def lifespan(app: FastAPI):
One shared instance is safe under concurrent requests. See
[examples/fastapi_app.py](examples/fastapi_app.py) for a runnable app.

### Custom httpx client
### Custom httpx2 client

Pass your own `httpx.AsyncClient` for custom SSL, proxies, event hooks, or
Pass your own `httpx2.AsyncClient` for custom SSL, proxies, event hooks, or
`MockTransport` in tests:

```python
client = httpx.AsyncClient(verify="/path/to/ca.pem", timeout=60)
client = httpx2.AsyncClient(verify="/path/to/ca.pem", timeout=60)
async with aiopynetbox.api(url, token=token, client=client) as nb:
...
```

Per httpx convention, a client you pass in is yours to close: `aclose()`
Per httpx2 convention, a client you pass in is yours to close: `aclose()`
and the context manager only close clients the Api created itself, so
one client can safely back several Api instances.

Expand Down
8 changes: 4 additions & 4 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
[project]
name = "aiopynetbox"
version = "0.1.0"
description = "Fully async NetBox API client built on httpx"
version = "0.2.0"
description = "Fully async NetBox API client built on httpx2"
readme = "README.md"
authors = [
{ name = "Cody Halley" }
]
license = "Apache-2.0"
license-files = ["LICENSE"]
requires-python = ">=3.11"
keywords = ["netbox", "async", "asyncio", "httpx", "api", "client", "ipam", "dcim"]
keywords = ["netbox", "async", "asyncio", "httpx2", "api", "client", "ipam", "dcim"]
classifiers = [
"Development Status :: 4 - Beta",
"Framework :: AsyncIO",
Expand All @@ -25,7 +25,7 @@ classifiers = [
"Typing :: Typed",
]
dependencies = [
"httpx>=0.28.1",
"httpx2>=2.0.0",
]

[project.urls]
Expand Down
4 changes: 2 additions & 2 deletions scripts/generate_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
from pathlib import Path
from typing import Any

import httpx
import httpx2

DEFAULT_URL = "https://demo.netbox.dev/api/schema/"
SRC = Path(__file__).resolve().parent.parent / "src" / "aiopynetbox"
Expand Down Expand Up @@ -181,7 +181,7 @@ def endpoint_stub(app: str, attr: str, filters: list[str], fields: list[str]) ->

def main() -> None:
url = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_URL
resp = httpx.get(
resp = httpx2.get(
url,
params={"format": "json"},
headers={"Accept": "application/json"},
Expand Down
16 changes: 8 additions & 8 deletions src/aiopynetbox/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from types import TracebackType
from typing import Any, Self

import httpx
import httpx2

from aiopynetbox.app import PluginsApp
from aiopynetbox.apps_generated import (
Expand Down Expand Up @@ -67,7 +67,7 @@ class Api:
Retry-After); transient 502/503/504 and connection
failures retry for GETs only, since an ambiguous write may
have been processed. 0 disables.
client: Custom httpx.AsyncClient (SSL config, proxies, mock
client: Custom httpx2.AsyncClient (SSL config, proxies, mock
transports). A supplied client is yours to close; the Api
closes only clients it creates itself.
"""
Expand All @@ -81,7 +81,7 @@ def __init__(
max_concurrency: int = 4,
pagination: str = "offset",
retries: int = 3,
client: httpx.AsyncClient | None = None,
client: httpx2.AsyncClient | None = None,
) -> None:
if pagination not in ("offset", "cursor"):
raise ValueError("pagination must be 'offset' or 'cursor'")
Expand All @@ -98,7 +98,7 @@ def __init__(
self._client = (
client
if client is not None
else httpx.AsyncClient(timeout=timeout, follow_redirects=True)
else httpx2.AsyncClient(timeout=timeout, follow_redirects=True)
)

self.circuits = CircuitsApp(self, "circuits")
Expand Down Expand Up @@ -128,7 +128,7 @@ async def aclose(self) -> None:
"""Close the connection pool, if this Api created it.

A client passed in via `client=` is the caller's to close
(httpx convention), so sharing one client across Api instances
(httpx2 convention), so sharing one client across Api instances
is safe.
"""
if self._owns_client:
Expand Down Expand Up @@ -159,7 +159,7 @@ async def _request_response(
params: dict[str, Any] | None = None,
json: Any = None,
headers: dict[str, str] | None = None,
) -> httpx.Response:
) -> httpx2.Response:
merged = {
"Accept": "application/json",
**self._auth_headers(),
Expand All @@ -172,7 +172,7 @@ async def _request_response(
resp = await self._client.request(
method, url, params=params, json=json, headers=merged
)
except httpx.TransportError:
except httpx2.TransportError:
# An ambiguous failure is only safely repeatable for GETs:
# a timed-out write may have been processed server-side.
if method != "GET" or attempt >= self.retries:
Expand All @@ -199,7 +199,7 @@ async def _request_response(
attempt += 1

@staticmethod
def _decode(resp: httpx.Response) -> Any:
def _decode(resp: httpx2.Response) -> Any:
try:
return resp.json()
except ValueError:
Expand Down
8 changes: 4 additions & 4 deletions src/aiopynetbox/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,13 @@

from __future__ import annotations

import httpx
import httpx2


class RequestError(Exception):
"""NetBox returned a non-success HTTP response."""

def __init__(self, response: httpx.Response) -> None:
def __init__(self, response: httpx2.Response) -> None:
self.response = response
self.status_code = response.status_code
self.url = str(response.url)
Expand All @@ -28,7 +28,7 @@ class AllocationError(Exception):
"""NetBox returned 409 Conflict for an allocation request
(e.g. available-ips with no room left)."""

def __init__(self, response: httpx.Response) -> None:
def __init__(self, response: httpx2.Response) -> None:
self.response = response
self.url = str(response.url)
self.error = "The requested allocation could not be fulfilled."
Expand All @@ -38,7 +38,7 @@ def __init__(self, response: httpx.Response) -> None:
class ContentError(Exception):
"""A successful response contained non-JSON content."""

def __init__(self, response: httpx.Response) -> None:
def __init__(self, response: httpx2.Response) -> None:
self.response = response
self.url = str(response.url)
self.error = (
Expand Down
Loading