Clients and nodes¶

Niwaki and AsyncNiwaki own the session; NiwakiNode / AsyncNiwakiNode are the DN-scoped handles they hand out. The clients observe — writing goes through the design DSL (The design DSL).

Clients¶

class niwaki.Niwaki( host=None, username=None, password=None, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, retry=None, token_cache=None, private_key=None, cert_dn=None, )[source]¶

Bases: object

Sync entry point for the ACI SDK.

Use as a sync context manager — authentication happens on __enter__ and the session is closed on __exit__. For async code use AsyncNiwaki instead.

Use as a context manager (recommended):

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    bds = aci.query(fvBD).fetch()

Use connect() for explicit lifecycle management:

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
try:
    bd = aci.root.tenant("prod").bd("web").read()
finally:
    aci.close()

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 (system CA store), a path to a PEM CA bundle (private/enterprise CA), or False (lab only). Default: True.
timeout (float) – HTTP request timeout in seconds. Default: 30.
refresh_threshold (int) – Seconds before token expiry at which a proactive refresh is triggered. Default: 60.
retry (RetryConfig | None) – Custom retry policy. Defaults to 3 attempts with exponential back-off. Construct with RetryConfig.

classmethod connect( host, username, password, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, retry=None, )[source]¶

Create a Niwaki instance and authenticate immediately (sync).

Equivalent to entering Niwaki(...) as a sync context manager but without the with statement — the same session construction and login path is used, so every constructor option (including retry) behaves identically.

Parameters:

host (str) – Base URL of the APIC.
username (str) – APIC username.
password (str) – APIC password.
verify_ssl (bool | str) – TLS verification — True, a PEM CA bundle path, or False. Default: True.
timeout (float) – HTTP timeout in seconds. Default: 30.
refresh_threshold (int) – Proactive refresh threshold in seconds. Default: 60.
retry (RetryConfig | None) – Custom retry policy. Defaults to 3 attempts with exponential back-off.

Returns:

Authenticated Niwaki instance.

Raises:

LoginError – APIC rejected the credentials.
ConnectionError – APIC host is unreachable.
TimeoutError – Login request timed out.
TLSError – TLS certificate verification failed.

Return type:

Niwaki

Example:

aci = Niwaki.connect(
    "https://sandboxapicdc.cisco.com",
    "admin",
    "ciscopsdt",
    verify_ssl=False,
)

close()[source]¶

Close the sync HTTP session and release network resources.

Called automatically on __exit__. Safe to call explicitly when not using Niwaki as a context manager. After close(), any further operation raises AuthError (mirrors AsyncNiwaki.close()).

property retry: RetryConfig | None¶

Custom retry policy passed at construction time, or None for the default.

Returns:: The RetryConfig override, or None when the session default (3 attempts) is in use.

classmethod with_client(client, host=None, username=None, password=None, **kwargs)[source]¶

Build a client that talks through an httpx client you configured.

Everything this SDK does not model — an outbound proxy, mutual TLS, a pinned CA bundle, a custom transport, per-route timeouts — is already expressible in httpx. Rather than mirror each as a parameter, this hands the whole client over; the returned object is identical in every other respect, and closing it closes the client you gave.

Parameters:

client (Client) – A configured httpx.Client. Pass the same host you gave it, since cookies and WebSocket URLs are derived from it.
host (str | None) – APIC base URL, or None to read APIC_HOST.
username (str | None) – Login name, or None to read APIC_USERNAME.
password (str | None) – Password, or None to read APIC_PASSWORD.
**kwargs (Any) – Any other constructor argument, validated as usual.

Returns:

A client using client for every request.

Return type:

Niwaki

Example:

behind_proxy = httpx.Client(base_url=host, proxy="http://proxy:3128")
aci = Niwaki.with_client(behind_proxy, host, user, password)

property apic_version: str | None¶

The firmware the connected controller stated at login.

Compare it against niwaki.catalog.schema_version() — the firmware this SDK’s models and vocabulary were generated from — to know whether you are inside the envelope it was built for. A mismatch is not a failure: reads stay tolerant, writes fail loudly. It is a reason to pilot rather than assume.

Returns:: The version string, or None before the client has connected.

property subscriptions: SubscriptionManager¶

Bulk introspection/management for this session’s object-subscriptions.

Returns:: A SubscriptionManager over the active session.

Example:

with Niwaki(...) as aci:
    sub = aci.query(fvBD).subscribe()
    for info in aci.subscriptions.list():
        print(info.path, info.is_stale)
    aci.subscriptions.close_all()

property root: NiwakiNode[ManagedObject]¶

Entry point for the ACI object hierarchy.

Returns a NiwakiNode at DN "uni" — the root of all ACI objects.

Returns:: NiwakiNode with dn = "uni".

Example:

bd = aci.root.tenant("prod").bd("web").read()

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

Access a node by explicit DN.

Parameters:

dn (str) – Full Distinguished Name (e.g. "uni/tn-prod/BD-web").
cls (type[T]) – ACI class hint. Defaults to ManagedObject.

Returns:

NiwakiNode scoped to the given DN.

Return type:

NiwakiNode[T]

Example:

bd = aci.node("uni/tn-prod/BD-web", fvBD).read()

query(cls: type[T]) → Query[T][source]¶

query(cls: str) → Query[ManagedObject]

Build a global class query for the entire ACI fabric.

Returns a Query builder targeting every instance of cls across the fabric (/api/class/{cls}.json). Add filters, scope constraints, or response enrichment before executing.

Also accepts a plain string class name for querying any of the ~15 000 ACI classes — including read-only operational and stats classes that are not in the generated set. Attributes on unregistered classes are accessible via obj.model_extra["attr"] or directly as obj.attr_name (Pydantic extra="allow").

Parameters:: cls (type[T] | str) – ACI class type (e.g. fvBD) or plain string class name (e.g. "topSystem").
Returns:: Query targeting the global class index.
Return type:: Query[T]

Example:

with Niwaki(...) as aci:
    # All tenants
    tenants = aci.query(fvTenant).fetch()

    # Filtered
    bds = aci.query(fvBD).where(arpFlood=True).fetch()

    # Scoped to a DN
    bds = aci.query(fvBD).under("uni/tn-prod").fetch()

    # Count only
    n = aci.query(fvBD).count()

    # Unregistered / read-only class
    nodes = aci.query("topSystem").naming_only().fetch()

class niwaki.AsyncNiwaki( host=None, username=None, password=None, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, max_concurrent=10, retry=None, token_cache=None, private_key=None, cert_dn=None, )[source]¶

Bases: object

Async entry point for the ACI SDK.

Mirrors Niwaki for async code. Use as an async context manager — authentication happens on __aenter__ and the session is closed on __aexit__.

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 (system CA store), a path to a PEM CA bundle (private/enterprise CA), or False (lab only). Default: True.
timeout (float) – HTTP request 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 simultaneous HTTP requests in flight. Default: 10.
retry (RetryConfig | None) – Custom retry policy. Defaults to 3 attempts with exponential back-off. Construct with RetryConfig.

Example:

async with AsyncNiwaki("https://apic.example.com", "admin", "pass") as aci:
    tenants = await aci.query(fvTenant).fetch()
    bd = await aci.root.tenant("prod").bd("web").read()

# Concurrent reads via gather()
async with AsyncNiwaki("https://apic.example.com", "admin", "pass") as aci:
    tenants, bd = await aci.gather(
        aci.query(fvTenant).fetch(),
        aci.root.tenant("prod").bd("web").read(),
    )

async classmethod connect( host, username, password, *, verify_ssl=True, timeout=30.0, refresh_threshold=60, max_concurrent=10, retry=None, )[source]¶

Create an AsyncNiwaki instance and authenticate immediately (async).

The async twin of Niwaki.connect() — for long-lived services that manage the lifecycle themselves rather than through async with. The same session construction and login path as the context manager is used, so every constructor option behaves identically. Pair with close() on shutdown.

Parameters:

host (str) – Base URL of the APIC.
username (str) – APIC username.
password (str) – APIC password.
verify_ssl (bool | str) – TLS verification — True, a PEM CA bundle path, or False. Default: True.
timeout (float) – HTTP timeout in seconds. Default: 30.
refresh_threshold (int) – Proactive refresh threshold in seconds. Default: 60.
max_concurrent (int) – Maximum simultaneous in-flight requests. Default: 10.
retry (RetryConfig | None) – Custom retry policy. Defaults to 3 attempts with exponential back-off.

Returns:

Authenticated AsyncNiwaki instance.

Raises:

LoginError – APIC rejected the credentials.
ConnectionError – APIC host is unreachable.
TimeoutError – Login request timed out.
TLSError – TLS certificate verification failed.

Return type:

AsyncNiwaki

Example:

aci = await AsyncNiwaki.connect(
    "https://sandboxapicdc.cisco.com",
    "admin",
    "ciscopsdt",
    verify_ssl=False,
)
try:
    bds = await aci.query("fvBD").fetch()
finally:
    await aci.close()

async close()[source]¶

Close the async HTTP session and release network resources.

Called automatically on __aexit__. Safe to call explicitly when not using AsyncNiwaki as an async context manager.

property retry: RetryConfig | None¶

Custom retry policy passed at construction time, or None for the default.

Returns:: The RetryConfig override, or None when the session default (3 attempts) is in use.

classmethod with_client(client, host=None, username=None, password=None, **kwargs)[source]¶

Build a client that talks through an httpx client you configured.

Parameters:

client (AsyncClient) – A configured httpx.AsyncClient. Pass the same host you gave it, since cookies and WebSocket URLs are derived from it.
host (str | None) – APIC base URL, or None to read APIC_HOST.
username (str | None) – Login name, or None to read APIC_USERNAME.
password (str | None) – Password, or None to read APIC_PASSWORD.
**kwargs (Any) – Any other constructor argument, validated as usual.

Returns:

A client using client for every request.

Return type:

AsyncNiwaki

Example:

behind_proxy = httpx.Client(base_url=host, proxy="http://proxy:3128")
aci = Niwaki.with_client(behind_proxy, host, user, password)

property apic_version: str | None¶

The firmware the connected controller stated at login.

Returns:: The version string, or None before the client has connected.

property max_concurrent: int¶

Upper bound on simultaneous in-flight HTTP requests for this client.

A staged push inherits this as its own fan-out bound, so push(mode="staged") runs at most this many operations of a wave at once. push(..., max_concurrent=n) can lower it further but never raise it above this number — construct the client with a higher limit to go wider.

Returns:: The limit set at construction (default 10).

property subscriptions: AsyncSubscriptionManager¶

Bulk introspection/management for this session’s object-subscriptions.

Returns:: An AsyncSubscriptionManager over the active session.

Example:

async with AsyncNiwaki(...) as aci:
    sub = await aci.query(fvBD).subscribe()
    for info in aci.subscriptions.list():
        print(info.path, info.is_stale)
    await aci.subscriptions.close_all()

property root: AsyncNiwakiNode[ManagedObject]¶

Entry point for the ACI object hierarchy.

Returns:: AsyncNiwakiNode at DN "uni".

Example:

bd = await aci.root.tenant("prod").bd("web").read()

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

Access a node by explicit DN.

Parameters:

dn (str) – Full Distinguished Name (e.g. "uni/tn-prod/BD-web").
cls (type[T]) – ACI class hint. Defaults to ManagedObject.

Returns:

AsyncNiwakiNode scoped to the given DN.

Return type:

AsyncNiwakiNode[T]

query(cls: type[T]) → AsyncQuery[T][source]¶

query(cls: str) → AsyncQuery[ManagedObject]

Build a global class query for the entire ACI fabric (async variant).

Mirrors query() for async contexts. The accumulator methods are synchronous; only the execution methods (fetch, first, count, stream) are coroutines, which means they compose naturally with gather().

Parameters:: cls (type[T] | str) – ACI class type (e.g. fvBD) or plain string class name.
Returns:: AsyncQuery targeting the global class index.
Return type:: AsyncQuery[T]

Example:

async with AsyncNiwaki(...) as aci:
    # All tenants
    tenants = await aci.query(fvTenant).fetch()

    # Concurrent reads via gather()
    tenants, bds = await aci.gather(
        aci.query(fvTenant).fetch(),
        aci.root.tenant("prod").bd().fetch(),
    )

    # Async streaming
    async for bd in aci.query(fvBD).stream():
        await process(bd)

async gather(*coros)[source]¶

Run multiple coroutines concurrently under a structured TaskGroup.

The APIC concurrency semaphore (max_concurrent) is already applied at the session level, so each coroutine naturally respects the limit. On the first coroutine that raises, TaskGroup cancels the others and raises an ExceptionGroup (Python 3.11+). Do not use gather for concurrent writes: a failure cancels the in-flight siblings, which can leave a second design partially applied.

Parameters:: *coros (Coroutine[Any, Any, Any]) – Coroutines to run concurrently (e.g. results of .read(), aci.query(...).fetch(), config.push(aci)).
Returns:: Tuple of results in the same order as the input coroutines.
Raises:: ExceptionGroup – If one or more coroutines raise. Remaining siblings are cancelled (TaskGroup semantics), not awaited.
Return type:: tuple[Any, …]

Example:

tenants, bd = await aci.gather(
    aci.query(fvTenant).fetch(),
    aci.root.tenant("prod").bd("web").read(),
)

Nodes¶

class niwaki.NiwakiNode(niwaki, dn, cls)[source]¶

Bases: _JargonNavMixin, Generic

A DN-scoped handle for observing a single ACI object (sync).

Created via Niwaki.root or Niwaki.node(). Navigate the ACI hierarchy with mo() or the vocabulary accessors — each call descends one level and computes the child DN automatically from the parent DN and the child’s RN. Terminal operations are read-side (read(), query()) plus delete(); configuration goes through niwaki.design.

Parameters:

niwaki (Niwaki) – Parent Niwaki instance.
dn (str) – Full Distinguished Name of this node (e.g. "uni/tn-prod/BD-web").
cls (type[T]) – ACI class associated with this node. Used to deserialise reads.

property dn: str¶

Full Distinguished Name of this node (read-only).

Returns:: DN string, e.g. "uni/tn-prod/BD-web".

Example:

node = aci.root.mo(fvTenant, name="prod").mo(fvBD, name="web")
node.dn  # → "uni/tn-prod/BD-web"

property cls: type[T]¶

ACI class associated with this node.

Returns:: The ManagedObject subclass used to construct and deserialise objects at this DN.

mo(cls, **naming_kwargs)[source]¶

Descend to a child node, computing its DN automatically.

Creates a temporary instance of cls purely to derive the RN from _rn_format. The child DN is self.dn + "/" + child.rn.

Parameters:

cls (type[U]) – ACI class of the child object.
**naming_kwargs (Any) – Naming props required by cls (e.g. name="web").

Returns:

A new NiwakiNode scoped to the child DN.

Raises:

ValidationError – If naming_kwargs violate the model constraints for cls.

Return type:

NiwakiNode[U]

Example:

aci.root                .mo(fvTenant, name="prod")                .mo(fvBD, name="web")                .mo(fvSubnet, ip="10.0.0.1/24")

read()[source]¶

Fetch the ACI object at this DN from the APIC.

Returns:

Typed ManagedObject instance.

Raises:

NotFoundError – The object does not exist on the APIC.
ForbiddenError – Insufficient privileges.
ServerError – APIC server-side error.

Return type:

Example:

bd = aci.root.mo(fvTenant, name="prod").mo(fvBD, name="web").read()
print(bd.unicast_routing)

delete()[source]¶

Delete the ACI object at this DN.

Raises:

NotFoundError – The object does not exist on the APIC.
ForbiddenError – Insufficient privileges.
ServerError – APIC server-side error.

Example:

aci.root.mo(fvTenant, name="prod").mo(fvBD, name="web").delete()

query(cls: type[U]) → Query[U][source]¶

query(cls: str) → Query[ManagedObject]

Build a query scoped to this node’s DN.

Returns all instances of cls that are descendants of this node, using the APIC query-target=subtree&target-subtree-class=cls query mode. Chain filter, scope, and enrichment methods before executing with fetch(), first(), count(), or stream().

This is also the entry point for querying unregistered classes (the full APIC class catalogue of ~15 000 classes): pass a plain string class name and the result will be a Query[ManagedObject] whose objects expose all APIC attributes via model_extra.

Parameters:: cls (type[U] | str) – ACI class type (e.g. fvBD) or plain string class name (e.g. "topSystem").
Returns:: Query scoped to this node’s DN.
Return type:: Query[U]

Example:

# All BDs under tenant "prod"
bds = aci.root.tenant("prod").query(fvBD).fetch()

# Count subnets in a specific BD
n = aci.root.tenant("prod").bd("web").query(fvSubnet).count()

# Unregistered / read-only class, scoped under this node's subtree
faults = aci.root.tenant("prod").query("faultInst").fetch()

class niwaki.AsyncNiwakiNode(niwaki, dn, cls)[source]¶

Bases: _JargonNavMixin, Generic

A DN-scoped handle for observing a single ACI object (async).

Mirrors NiwakiNode for async contexts. Terminal operations (read(), delete()) are coroutines; configuration goes through niwaki.design.

Parameters:

niwaki (AsyncNiwaki) – Parent AsyncNiwaki instance.
dn (str) – Full Distinguished Name of this node.
cls (type[T]) – ACI class associated with this node. Used to deserialise reads.

property dn: str¶

Full Distinguished Name of this node (read-only).

Returns:: DN string, e.g. "uni/tn-prod/BD-web".

property cls: type[T]¶

ACI class associated with this node.

Returns:: The ManagedObject subclass used to construct and deserialise objects at this DN.

mo(cls, **naming_kwargs)[source]¶

Descend to a child node, computing its DN automatically.

Parameters:

cls (type[U]) – ACI class of the child object.
**naming_kwargs (Any) – Naming props required by cls.

Returns:

A new AsyncNiwakiNode scoped to the child DN.

Raises:

ValidationError – If naming_kwargs violate model constraints.

Return type:

AsyncNiwakiNode[U]

async read()[source]¶

Fetch the ACI object at this DN from the APIC.

Returns:

Typed ManagedObject instance.

Raises:

NotFoundError – The object does not exist.
ForbiddenError – Insufficient privileges.
ServerError – APIC server-side error.

Return type:

async delete()[source]¶

Delete the ACI object at this DN.

Raises:

NotFoundError – Object does not exist.
ForbiddenError – Insufficient privileges.
ServerError – APIC server-side error.

query(cls: type[U]) → AsyncQuery[U][source]¶

query(cls: str) → AsyncQuery[ManagedObject]

Build a query scoped to this node’s DN (async variant).

Mirrors query() for async contexts. All accumulator methods are synchronous; only the execution methods (fetch, first, count, stream) are coroutines.

Parameters:: cls (type[U] | str) – ACI class type or plain string class name.
Returns:: AsyncQuery scoped to this node’s DN.
Return type:: AsyncQuery[U]

Example:

bds = await aci.root.tenant("prod").query(fvBD).fetch()
n   = await aci.root.tenant("prod").bd("web").query(fvSubnet).count()

Subscription management¶

aci.subscriptions — bulk introspection/refresh/stop over every subscription open on the session’s shared WebSocket; see Subscribing to live changes.

class niwaki.facade.SubscriptionManager(session)[source]¶

Bases: object

Bulk introspection/management for a session’s object-subscriptions (sync).

Reached via subscriptions. Operates on whichever subscriptions are currently open on the session’s shared WebSocket — every method is a safe no-op (empty list, or nothing to do) until at least one aci.query(...).subscribe() call has opened it.

list()[source]¶

List every subscription currently tracked, or [] if none is open.

Returns:: One SubscriptionInfo per tracked subscription — .is_stale flags one with at least one recent refresh failure.
Return type:: list[SubscriptionInfo]

Example:

for info in aci.subscriptions.list():
    print(info.path, info.consecutive_refresh_failures, info.is_stale)

refresh_all()[source]¶

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

A diagnostic tool, distinct from the automatic background refresh sweep — a failure here never triggers the automatic recovery escalation a missed scheduled refresh would.

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

close_all()[source]¶

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

Every open Subscription’s iterator ends with a plain StopIteration. Distinct from closing the whole Niwaki session: a later aci.query(...).subscribe() reuses the same still-open socket immediately, with no reconnect needed.

class niwaki.facade.AsyncSubscriptionManager(session)[source]¶

Bases: object

Async counterpart of SubscriptionManager. Reached via subscriptions.

list()[source]¶

List every subscription currently tracked, or [] if none is open.

Synchronous even on the async facade — see list_subscriptions().

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

async refresh_all()[source]¶

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

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

async close_all()[source]¶

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

Vocabulary navigation¶

Nodes — and the clients themselves — resolve operator vocabulary into typed child handles: aci.tenant("prod").bd("web") is not a hardcoded method chain, it is resolved against the APIC containment model at each step.

NiwakiNode.__getattr__(attr)¶

Resolve vocabulary attribute names to typed child navigators.

Parameters:: attr (str) – Method name to resolve (e.g. "tenant", "bd").
Returns:: Callable that, when invoked with naming kwargs, returns a child node of the appropriate concrete type (sync or async).
Raises:: AttributeError – attr is not a known child of this node’s class.
Return type:: Any

Retry policy¶

Retries are configured on the client with RetryConfig (documented with the transport layer that applies it).