Transport¶

Transport layer — APIC sessions, retry policy, and the transport boundary.

The clients (Niwaki, AsyncNiwaki) own a session and hand it to the design engine and the query builder through two pairs of structural protocols — MoWriter / AsyncMoWriter (write) and MoReader / AsyncMoReader (read). Anything that implements them is a valid transport, which is what makes the SDK testable without a fabric (see the Testing your automation guide).

Sessions are managed by the clients; construct one directly only when you need a transport without the facade.

The transport boundary¶

The facade and the push engine depend on these structural protocols, never on the concrete sessions — any conforming object is a valid transport, which is how you test your automation without a fabric (Testing your automation).

class niwaki.transport.MoWriter(*args, **kwargs)[source]¶

Bases: Protocol

Structural type for synchronous ACI write transports.

post_mo(dn, payload)[source]¶

POST an APIC envelope to the given DN.

Parameters:

dn (str) – Full Distinguished Name of the target object.
payload (dict[str, Any]) – APIC envelope dict ({"fvBD": {"attributes": {...}}}).

delete_mo(dn)[source]¶

DELETE the ACI object at the given DN.

Parameters:: dn (str) – Full Distinguished Name of the target object.

class niwaki.transport.MoReader(*args, **kwargs)[source]¶

Bases: Protocol

Structural type for synchronous typed single-MO reads.

get_mo(dn, cls)[source]¶

Fetch one MO by DN, typed as cls.

Raises:: NotFoundError – No object exists at dn.

class niwaki.transport.AsyncMoWriter(*args, **kwargs)[source]¶

Bases: Protocol

Structural type for asynchronous ACI write transports.

Example — minimal test stub:

class FakeSession:
    async def post_mo(self, dn: str, payload: dict[str, Any]) -> None:
        print(f"POST {dn}")

    async def delete_mo(self, dn: str) -> None:
        print(f"DELETE {dn}")

async post_mo(dn, payload)[source]¶

POST an APIC envelope to the given DN.

Parameters:

dn (str) – Full Distinguished Name of the target object.
payload (dict[str, Any]) – APIC envelope dict ({"fvBD": {"attributes": {...}}}).

async delete_mo(dn)[source]¶

DELETE the ACI object at the given DN.

Parameters:: dn (str) – Full Distinguished Name of the target object.

class niwaki.transport.AsyncMoReader(*args, **kwargs)[source]¶

Bases: Protocol

Structural type for asynchronous typed single-MO reads.

async get_mo(dn, cls)[source]¶

Fetch one MO by DN, typed as cls.

Raises:: NotFoundError – No object exists at dn.

Sessions¶

Authentication, proactive token refresh, retries and transparent pagination live here. The clients construct and close a session for you — reach for these classes only when you want a transport without the facade.

class niwaki.transport.ApicSession( host=None, username=None, password=None, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, retry=RetryConfig(attempts=3, wait_initial=0.5, wait_max=5.0, wait_jitter=0.5), )[source]¶

Bases: object

Synchronous APIC session with automatic token management.

Handles login, proactive token refresh, and transparent re-authentication. Designed for use as a context manager or standalone.

Authentication and token lifecycle:

with ApicSession("https://apic.example.com", "admin", "pass") as s:
    imdata = s.get("/api/class/fvTenant.json")

# Standalone usage:
s = ApicSession(host="https://apic.example.com")
s.login()
imdata = s.get("/api/mo/uni/tn-Prod.json")
s.close()

Environment variable fallbacks (used when arguments are not provided):: APIC_HOST : Base URL of the APIC. APIC_USERNAME : APIC username. APIC_PASSWORD : APIC password.

Parameters:

host (str | None) – Base URL of the APIC (e.g. "https://sandboxapicdc.cisco.com"). Falls back to APIC_HOST environment variable if omitted.
username (str | None) – APIC username. Falls back to APIC_USERNAME if omitted.
password (str | None) – APIC password. Falls back to APIC_PASSWORD if omitted.
verify_ssl (bool | str) – TLS verification — True verifies against the system CA store, a path to a PEM CA bundle verifies against a private CA, False disables verification. Keep False for APICs with self-signed certificates (not recommended in production). Default: True.
timeout (float) – HTTP timeout in seconds (connect + read). Default: 30.
refresh_threshold (int) – Seconds before token expiry at which a proactive refresh is triggered. Default: 60.

Raises:

KeyError – If host, username, or password are omitted and the corresponding environment variables are not set.

Note

This implementation is not thread-safe: only the token refresh is lock-protected, not the underlying client state. For concurrent usage, create one session per thread — or use AsyncApicSession.

close()[source]¶

Close the httpx client and release network resources.

Call explicitly when not using the session as a context manager. After close(), any request will raise an httpx error. Also tears down the subscription WebSocket, if one was ever opened — every blocked subscription iterator wakes with a plain StopIteration.

property is_closed: bool¶

True after close() has been called.

Returns:: Whether the underlying httpx client has been closed.

property is_authenticated: bool¶

True once login() has succeeded and the token is valid.

Returns:: Whether the session holds a live authentication token.

property retry: RetryConfig¶

Active retry policy for this session.

Returns:: The RetryConfig in use.

login()[source]¶

Authenticate the session against the APIC via /api/aaaLogin.json.

Submits credentials, stores the returned token with its TTL, and sets the APIC-cookie cookie on the underlying HTTP client. Subsequent requests are automatically authenticated.

Raises:

LoginError – The APIC rejected the credentials or the response is malformed.
ConnectionError – The APIC host is unreachable.
TimeoutError – The login request exceeded the configured timeout.
TLSError – TLS verification failed (invalid certificate, etc.).

Example:

session = ApicSession("https://apic.example.com", "admin", "secret")
session.login()
# session._token_state now holds the token and its expiry

get(path, **params)[source]¶

Execute a GET against the APIC REST API and return the imdata list.

Ensures token validity before the request. Automatically retries on transient network errors (3 attempts, exponential backoff). Handles mid-session 401s by re-authenticating and replaying the request once.

Parameters:

path (str) – API path relative to the base URL (e.g. "/api/mo/uni/tn-MyTenant.json").
**params (Any) – Optional query string parameters (e.g. **{"query-target": "children", "rsp-subtree": "full"}).

Returns:

The imdata list from the APIC JSON response. Empty list if the APIC returns an empty object (totalCount: "0").

Raises:

AuthError – Not authenticated and automatic re-auth is not possible.
SessionExpiredError – Token expired and re-auth failed.
NotFoundError – HTTP 404 — the MO does not exist.
UnauthorizedError – HTTP 401 persisting after re-authentication.
ForbiddenError – HTTP 403 — insufficient privileges.
ServerError – HTTP 5xx — APIC server-side error.
ConnectionError – Host unreachable after all retry attempts.
TimeoutError – Timeout exceeded after all retry attempts.
TLSError – TLS verification error.

Return type:

list[dict[str, Any]]

Example:

with ApicSession("https://apic.example.com", "admin", "pass") as s:
    tenants = s.get("/api/class/fvTenant.json")
    for item in tenants:
        print(item["fvTenant"]["attributes"]["name"])

get_mo(dn, cls=<class 'niwaki.models.base.ManagedObject'>)[source]¶

Fetch a single MO by DN, typed as cls.

Part of the transport boundary (niwaki.transport._protocols.MoReader).

Parameters:

dn (str) – Full Distinguished Name of the object.
cls (type[_T]) – Model class used to deserialise the response.

Returns:

The typed instance.

Raises:

NotFoundError – No object exists at dn.

Return type:

subscribe(path, params, *, refresh_timeout=None)[source]¶

Subscribe to push notifications for a query, over the session’s shared WebSocket.

Part of the transport boundary (niwaki.transport._protocols.MoSubscriber). The APIC multiplexes every subscription for a session over one WebSocket, opened lazily on the first call to this method; a refresh sweep and reconnect-and-resubscribe are handled automatically in the background — a caller never hand-rolls either.

Parameters:

path (str) – API path relative to base URL, exactly as passed to get() (e.g. "/api/class/fvBD.json").
params (dict[str, str]) – Query string parameters (filters/scoping). subscription and refresh-timeout are added internally — do not include them here.
refresh_timeout (int | None) – Override the APIC’s default 60 s subscription timeout. The subscription refreshes itself automatically on a schedule derived from this value regardless.

Returns:

A RawSubscription — .initial for the synchronous snapshot, then iterate for live push items.

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
SubscribeRejectedError – The APIC rejected the subscribe request.

Return type:

RawSubscription

Example:

sub = session.subscribe("/api/class/fvBD.json", {})
for item in sub:
    print(item)

list_subscriptions()[source]¶

List every subscription currently tracked on this session’s socket.

Returns an empty list if no subscription was ever opened — this never opens the WebSocket itself.

Returns:: One SubscriptionInfo per tracked subscription.
Return type:: list[SubscriptionInfo]

refresh_all_subscriptions()[source]¶

Force an immediate refresh of every tracked subscription, on demand.

A no-op returning an empty list if no subscription was ever opened. See SubscriptionSocket’s method of the same name for the escalation-safety semantics.

Returns:: The post-refresh snapshot of every subscription.
Return type:: list[SubscriptionInfo]

close_all_subscriptions()[source]¶

Stop every tracked subscription — the shared socket itself stays open.

A no-op if no subscription was ever opened. Distinct from close(), which tears down the whole socket; see close_all_subscriptions().

post_mo(dn, payload)[source]¶

POST an APIC envelope to a Managed Object URL.

Used for both create and update operations. APIC applies upsert semantics — the object is created if absent or updated if it exists. Only the fields present in the payload are modified; unspecified fields retain their current values.

Parameters:

dn (str) – Full Distinguished Name of the target object (e.g. "uni/tn-prod/BD-web").
payload (dict[str, Any]) – APIC envelope dict as produced by to_apic().

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
ForbiddenError – HTTP 403 — insufficient privileges.
NotFoundError – HTTP 404 — invalid DN structure.
ServerError – HTTP 5xx — APIC server-side error.
ConnectionError – Network error after all retry attempts.
TimeoutError – Timeout exceeded.
TLSError – TLS verification error.

Example:

session.post_mo("uni/tn-prod/BD-web", bd.to_apic())

delete_mo(dn)[source]¶

DELETE a Managed Object by Distinguished Name.

Permanently removes the object and all its children from the APIC. This operation is irreversible.

Parameters:

dn (str) – Full Distinguished Name of the object to delete (e.g. "uni/tn-prod/BD-web").

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
NotFoundError – HTTP 404 — the object does not exist.
ForbiddenError – HTTP 403 — insufficient privileges.
ServerError – HTTP 5xx — APIC server-side error.
ConnectionError – Network error after all retry attempts.
TimeoutError – Timeout exceeded.
TLSError – TLS verification error.

Example:

session.delete_mo("uni/tn-prod/BD-web")

class niwaki.transport.AsyncApicSession( host=None, username=None, password=None, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, max_concurrent=10, retry=RetryConfig(attempts=3, wait_initial=0.5, wait_max=5.0, wait_jitter=0.5), )[source]¶

Bases: object

Asynchronous APIC session with automatic token management.

Mirrors ApicSession for use in async contexts. All public methods are coroutines. Concurrent requests are rate-limited by a asyncio.Semaphore; token refresh is protected by an asyncio.Lock so exactly one refresh runs at a time even when many coroutines call _ensure_token simultaneously.

Designed for use as an async context manager or standalone:

async with AsyncApicSession("https://apic.example.com", "admin", "pass") as s:
    data = await s.get("/api/class/fvTenant.json")

# Standalone:
s = AsyncApicSession(host="https://apic.example.com")
await s.login()
data = await s.get("/api/class/fvTenant.json")
await s.close()

Environment variable fallbacks (used when arguments are not provided):: APIC_HOST : Base URL of the APIC. APIC_USERNAME : APIC username. APIC_PASSWORD : APIC password.

Parameters:

host (str | None) – Base URL of the APIC (e.g. "https://sandboxapicdc.cisco.com"). Falls back to APIC_HOST if omitted.
username (str | None) – APIC username. Falls back to APIC_USERNAME if omitted.
password (str | None) – APIC password. Falls back to APIC_PASSWORD if omitted.
verify_ssl (bool | str) – TLS verification — True (system CA store), a path to a PEM CA bundle (private CA), or False. Default: True.
timeout (float) – HTTP timeout in seconds. Default: 30.
refresh_threshold (int) – Seconds before token expiry at which a proactive refresh is triggered. Default: 60.
max_concurrent (int) – Maximum number of simultaneous in-flight HTTP requests. APIC rejects excessive concurrent connections; this semaphore prevents that. Default: 10.

Raises:

KeyError – If any of host/username/password are omitted and the corresponding environment variables are not set.

Note

Not safe for concurrent use from multiple event loops. Each event loop should create its own session.

async close()[source]¶

Close the async HTTP client and release network resources.

Call explicitly when not using the session as an async context manager. After close(), any request will raise an httpx error. Also tears down the subscription WebSocket, if one was ever opened — every blocked subscription iterator ends with a plain StopAsyncIteration.

property is_closed: bool¶

True after close() has been called.

Returns:: Whether the underlying httpx async client has been closed.

property is_authenticated: bool¶

True once login() has succeeded and the token is valid.

Returns:: Whether the session holds a live authentication token.

property retry: RetryConfig¶

Active retry policy for this session.

Returns:: The RetryConfig in use.

async login()[source]¶

Authenticate against the APIC via /api/aaaLogin.json.

Submits credentials, stores the returned token with its TTL, and sets the APIC-cookie cookie on the underlying HTTP client.

Raises:

LoginError – The APIC rejected the credentials or the response is malformed.
ConnectionError – The APIC host is unreachable.
TimeoutError – The login request exceeded the configured timeout.
TLSError – TLS verification failed.

Example:

session = AsyncApicSession("https://apic.example.com", "admin", "pass")
await session.login()

async get(path, **params)[source]¶

Execute a GET against the APIC REST API and return the imdata list.

Ensures token validity, applies the concurrency semaphore, and retries on transient network errors. Handles mid-session 401s by re-authenticating and replaying once.

Parameters:

path (str) – API path relative to the base URL (e.g. "/api/class/fvTenant.json").
**params (Any) – Optional query string parameters.

Returns:

The imdata list from the APIC JSON response.

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
NotFoundError – HTTP 404.
UnauthorizedError – HTTP 401 persisting after re-auth.
ForbiddenError – HTTP 403.
ServerError – HTTP 5xx.
ConnectionError – Host unreachable after all retries.
TimeoutError – Timeout exceeded after all retries.
TLSError – TLS verification error.

Return type:

list[dict[str, Any]]

Example:

async with AsyncApicSession("https://apic.example.com", "admin", "pass") as s:
    tenants = await s.get("/api/class/fvTenant.json")

async get_mo(dn, cls=<class 'niwaki.models.base.ManagedObject'>)[source]¶

Fetch a single MO by DN, typed as cls.

Part of the transport boundary (niwaki.transport._protocols.AsyncMoReader).

Parameters:

dn (str) – Full Distinguished Name of the object.
cls (type[_T]) – Model class used to deserialise the response.

Returns:

The typed instance.

Raises:

NotFoundError – No object exists at dn.

Return type:

async subscribe(path, params, *, refresh_timeout=None)[source]¶

Subscribe to push notifications for a query, over the session’s shared WebSocket.

Part of the transport boundary (niwaki.transport._protocols.AsyncMoSubscriber). Mirrors niwaki.transport.ApicSession.subscribe(): the APIC multiplexes every subscription for a session over one WebSocket, opened lazily on the first call to this method; a refresh sweep and reconnect-and-resubscribe are handled automatically in the background — a caller never hand-rolls either.

Parameters:

path (str) – API path relative to base URL, exactly as passed to get() (e.g. "/api/class/fvBD.json").
params (dict[str, str]) – Query string parameters (filters/scoping). subscription and refresh-timeout are added internally — do not include them here.
refresh_timeout (int | None) – Override the APIC’s default 60 s subscription timeout. The subscription refreshes itself automatically on a schedule derived from this value regardless.

Returns:

A AsyncRawSubscription — .initial for the synchronous snapshot, then iterate (async for) for live push items.

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
SubscribeRejectedError – The APIC rejected the subscribe request.

Return type:

AsyncRawSubscription

Example:

sub = await session.subscribe("/api/class/fvBD.json", {})
async for item in sub:
    print(item)

list_subscriptions()[source]¶

List every subscription currently tracked on this session’s socket.

Returns an empty list if no subscription was ever opened — this never opens the WebSocket itself. Synchronous (matches the sync ApicSession accessor and the underlying lock-free read) even on this async session.

Returns:: One SubscriptionInfo per tracked subscription.
Return type:: list[SubscriptionInfo]

async refresh_all_subscriptions()[source]¶

Force an immediate refresh of every tracked subscription, on demand.

A no-op returning an empty list if no subscription was ever opened. See SubscriptionSocket’s method of the same name for the escalation-safety semantics.

Returns:: The post-refresh snapshot of every subscription.
Return type:: list[SubscriptionInfo]

async close_all_subscriptions()[source]¶

Stop every tracked subscription — the shared socket itself stays open.

A no-op if no subscription was ever opened. Distinct from close(), which tears down the whole socket; see close_all_subscriptions().

async post_mo(dn, payload)[source]¶

POST an APIC envelope to a Managed Object URL.

APIC applies upsert semantics — the object is created if absent or updated if it exists.

Parameters:

dn (str) – Full Distinguished Name of the target object.
payload (dict[str, Any]) – APIC envelope dict as produced by to_apic().

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
ForbiddenError – HTTP 403.
NotFoundError – HTTP 404 — invalid DN structure.
ServerError – HTTP 5xx.
ConnectionError – Network error after all retries.
TimeoutError – Timeout exceeded.
TLSError – TLS verification error.

Example:

await session.post_mo("uni/tn-prod/BD-web", bd.to_apic())

async delete_mo(dn)[source]¶

DELETE a Managed Object by Distinguished Name.

Parameters:

dn (str) – Full Distinguished Name of the object to delete.

Raises:

AuthError – Not authenticated.
SessionExpiredError – Token expired and re-auth failed.
NotFoundError – HTTP 404.
ForbiddenError – HTTP 403.
ServerError – HTTP 5xx.
ConnectionError – Network error after all retries.
TimeoutError – Timeout exceeded.
TLSError – TLS verification error.

Example:

await session.delete_mo("uni/tn-prod/BD-web")

Retry policy¶

class niwaki.transport.RetryConfig(attempts=3, wait_initial=0.5, wait_max=5.0, wait_jitter=0.5)[source]¶

Bases: object

Stamina retry parameters for APIC HTTP transport.

A frozen (immutable, hashable) value object. All retry parameters are forwarded verbatim to stamina.retry_context / stamina.retry for both GET and mutating requests.

Parameters:

attempts (int) – Total number of attempts (first try + retries). 1 effectively disables retries. Default: 3.
wait_initial (float) – Initial backoff in seconds before the first retry. Default: 0.5.
wait_max (float) – Maximum backoff in seconds (exponential backoff is capped here). Default: 5.0.
wait_jitter (float) – Random jitter added to each backoff in seconds to prevent thundering-herd effects. Default: 0.5.

Example:

# Disable retries entirely (one attempt, no backoff)
no_retry = RetryConfig(attempts=1)

# Production: three attempts with up to 10 s max backoff
prod = RetryConfig(attempts=3, wait_max=10.0)