YaPyAci

Yet Another Python ACI (aka YaPiAci) is a module that helps making queries against Cisco ACI using the REST API, as well as handling the response in a very simple way.

The APIC REST API answers with deeply nested JSON. YaPyAci wraps that response into objects you can navigate with the dot notation, search at any depth, and compare — and takes care of the plumbing around it: authentication and token refresh, failover across the members of an APIC cluster, event subscriptions and the WebSocket event stream.

• Installation
• Quick start
• Contexts and clients
• Connecting
• Querying
• Working with results
• Subscribing to events
• Exceptions
• Logging
• Notes and limitations
• License

Installation

pip install yapyaci

Or from a clone of this repository:

pip install .

Requires Python 3.9 or later. The only dependency is websocket-client 1.0 or later.

Running the tests

The test suite lives in tests/ and is not shipped in the wheel; clone the repository to run it. It needs pytest, which comes with the dev extra, and it runs entirely offline - there is no APIC involved at any point. The same extra brings ruff, which the repository is clean under.

python3 -m venv .venv
.venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest -q
.venv/bin/ruff check

Both run in CI, on 3.8 through 3.13, along with a check that the built distribution is sound and that the suite passes from the unpacked sdist.

The JSON fixtures the tests read are generated by tests/make_fixtures.py from a configuration export that is not part of the repository, so regenerating them is only needed when that export changes.

Quick start

from yapyaci import Api, ApiCtx

ctx = ApiCtx(ip="192.168.1.1", username="my_user", password="my_password")
api = Api(ctx)

# One object, by distinguished name
tenant = api.lookup_by_dn("uni/tn-common")
print(tenant.att.name)                      # common

# All objects of a class
for tenant in api.lookup_by_class("fvTenant"):
    print(tenant.att.dn)                    # uni/tn-mgmt, uni/tn-infra, ...

# A whole subtree, searched at any depth
infra = api.lookup_by_dn("uni/tn-infra", rsp_subtree="full")
for bd in infra.search("fvBD", arpFlood="yes"):
    print(bd.att.dn)

Contexts and clients

A connection is split into a context, which holds what is shared, and a client, which sends the requests. Each comes in two flavours:

	Context	Client	Adds
Plain	ApiCtx	Api	Nothing. Queries, and no background activity whatsoever.
Long-running	EApiCtx	EApi	Keeps the session alive by itself, and can subscribe to events.

The context holds the APIC address(es), the credentials, the authentication token and its refresh timeout, and the health of each cluster member. The client is cheap: it owns a single HTTP connection and exposes the query methods.

Api or EApi?

EApi is Api — same constructor, same query methods, same results — plus two things it does on your behalf.

It keeps the token alive. APIC hands out a token with a lifetime (refreshTimeoutSeconds, quoted at login). Api never refreshes it. That is not fatal, because an expired token makes APIC answer 403, and YaPyAci clears the token on a 403 so the next query logs in again — but the query that got the 403 still raises QueryFailure, and is not retried for you. On a context that lives longer than the token, plain Api therefore fails one query per token lifetime. The first lookup made through EApi starts a thread that refreshes the token a minute before it expires, so it never gets that far.

It can subscribe to events. subscribe(), start_stream() and stop_stream() exist only on EApi.

Which to use:

Situation	Use
A script, a CLI tool, anything shorter-lived than the token	Api
A poller or daemon holding a context open for hours, even with no subscriptions	EApi
Anything that subscribes to events or opens the WebSocket	EApi

EApi requires an EApiCtx; Api accepts either. So a long-running program that also fires one-off queries can build both clients on one EApiCtx and share a single login:

from yapyaci import Api, EApi, EApiCtx

ctx = EApiCtx(ip="192.168.1.1", username="my_user", password="my_password")

poller = EApi(ctx)                          # refreshes the token for the whole context
helper = Api(ctx)                           # rides on the same token

Sharing a context across threads

A context can be shared by any number of clients. This is the intended way to query APIC from several threads: give each thread its own client, all built on one context. The context serialises the initial login behind a lock, so a single authentication is performed no matter how many threads start at once, and the resulting token is reused by all of them.

import threading, time
from yapyaci import EApi, EApiCtx

ctx = EApiCtx(ip="192.168.1.1", username="my_user", password="my_password")

def worker():
    api = EApi(ctx)                         # own connection, shared credentials and token
    while True:
        print(api.lookup_by_class("fabricHealthTotal")[0].att.cur)
        time.sleep(5)

for _ in range(5):
    threading.Thread(target=worker).start()

Background threads

EApi starts at most one of each per context, and they are shared by every client on it:

Thread	Started by	Stops when
Token refresh	The first EApi query, or start_stream()	The session on the context is closed, or a refresh fails.
Subscription refresh	The first subscribe()	The WebSocket closes, or a subscription fails to refresh — in which case it closes the WebSocket too.
WebSocket stream	start_stream()	stop_stream(), or the connection drops.

None of them is a daemon thread, so a program using EApi will not exit while they are running, even after the main thread finishes. To wind a session down, close the stream and drop the authentication:

api.stop_stream()                           # closes the WebSocket
ctx.session.close()                         # lets the token refresh thread finish

Both threads check whether they should keep going only after their sleep, so the process can linger for up to one refresh interval — under two minutes for the token thread, three for the subscription thread — before it exits.

Connecting

Authentication

Authentication is lazy: no request is sent when the context or the client is created. The first query logs in and stores the token on the context, and every client on that context reuses it. Bad credentials therefore raise AuthFailure on that first query rather than at construction time.

What the login returned lives on ctx.session, a Session: ctx.session.is_open says whether there is one, ctx.session.refresh_timeout is the lifetime APIC quoted, and ctx.session.close() drops it so that the next query logs in again.

The token has a limited lifetime. Whether it is refreshed before it expires is the main practical difference between the two clients — see Api or EApi?.

APIC clusters

ip accepts a single address or a list of them:

ctx = ApiCtx(ip=["192.168.1.1", "192.168.1.2", "192.168.1.3"],
             username="my_user",
             password="my_password")

Queries are spread over the cluster round-robin. When a member fails to answer — connection reset, timeout, or an answer that never claimed to be JSON, which is a load balancer or a proxy answering instead of a controller — it is poisoned: the request is immediately retried against the next member, and the failed one is skipped for the next 10 minutes. If every member ends up poisoned the list is cleared and they are all tried again. ConnFailure is raised only when no member could serve the request, and its message lists what each one answered.

A member that answers Content-Type: application/json with something that does not parse is a different case: it answered, so there is nothing to fail over to, and every other member would answer the same. That raises InvalidJsonPayload naming the member and the status code, and poisons nothing. APIC does produce such answers — the message of an mqapi2 400 holds unescaped quotes, so its own error body is not valid JSON.

The cluster is a HostPool, reachable as ctx.hosts. A member can be added to a live context at any time, and the pool can be asked what it currently avoids:

ctx.hosts.add("192.168.1.4")
ctx.hosts.hosts                             # every member, in the order they were added
ctx.hosts.poisoned                          # the ones being skipped right now

Closing

api.disconnect() closes the underlying HTTP connection. The client stays usable — the next query opens a new connection to one of the cluster members. The authentication token lives on the context and is not affected.

Querying

lookup_by_dn

Queries one managed object by its distinguished name (/api/mo/<dn>.json) and returns an AciTree.

tree = api.lookup_by_dn("uni/tn-common")

print(tree.cls)                             # fvTenant
print(tree.att.dn)                          # uni/tn-common
print(tree.att.name)                        # common

lookup_by_class

Queries every object of a class (/api/node/class/<cls>.json) and returns a ListOfAciTree, which iterates and indexes into AciTree objects.

ltree = api.lookup_by_class("fvTenant")

print(ltree.total_count)                    # 4
print(ltree[0].att.dn)                      # uni/tn-mgmt

for tree in ltree:
    print(tree.att.name)                    # mgmt, infra, common, my_tenant

Query filters

Both functions accept every query filter the Cisco REST API offers (see the Cisco APIC REST API Configuration Guide); they are passed straight through to the query string. As Python does not allow - in a keyword name, use an underscore instead — rsp-subtree becomes rsp_subtree, target-subtree-class becomes target_subtree_class, and so on.

# The tenant and its whole subtree
api.lookup_by_dn("uni/tn-common", rsp_subtree="full")

# The tenant and its direct children only
api.lookup_by_dn("uni/tn-common", rsp_subtree="children")

# Every bridge domain with unicast routing disabled
api.lookup_by_class("fvBD", query_target_filter='eq(fvBD.unicastRoute,"no")')

lookup

lookup() takes either a dn or a cls keyword plus the same filters, and returns the base AciSet without deciding for you whether the result is one tree or many. It is what subscriptions are built on, because the response carries a subscription_id. Passing neither dn nor cls raises InvalidLookupRequest.

result = api.lookup(cls="fvTenant", rsp_subtree="full")
ltree = result.ltree()                      # cast when you know what you got

mqapi2

mqapi2() reaches the APIC /mqapi2/ endpoints, the ones the GUI uses for its own tooling rather than the MIT. It takes the name of the endpoint plus its parameters, and returns a ListOfAciTree:

# GET /mqapi2/<tool>.json?<parameters>
api.mqapi2("<tool>", some_parameter="value")

Which endpoints exist and what they expect is not part of the documented REST API; the usual way to find out is to watch what the APIC GUI itself requests.

Working with results

Every query returns one of three classes, all sharing the same base:

Class	Returned by	Holds
AciSet	lookup()	The raw response. Searchable and castable, but no per-object accessors.
AciTree	lookup_by_dn()	Exactly one object and its subtree.
ListOfAciTree	lookup_by_class(), search(), children	Several trees. Iterable and indexable, yielding AciTree.

Object attributes

The attributes of an AciTree are reached through the att accessor, and its ACI class through cls:

tree = api.lookup_by_dn("uni/tn-common")

print(tree.cls)                             # fvTenant
print(tree.att.dn)                          # uni/tn-common
print(tree.att.descr)                       # ''

Asking for an attribute that APIC did not return raises AttributeError, naming both the attribute and the class, which is usually the quickest way to notice that a query needs rsp_subtree or a different filter. Being an AttributeError rather than something of its own, it also means hasattr(tree.att, "x") and getattr(tree.att, "x", default) work as they do on any other object.

tree.attributes is the same data as a plain dictionary, and it is the way to reach the attributes whose APIC name is not a usable Python name. A VLAN pool block is the one you meet first: fvnsEncapBlk carries from and to, and blk.att.from is a syntax error, not a lookup failure.

for blk in api.lookup_by_class("fvnsEncapBlk"):
    print(blk.attributes["from"], blk.attributes["to"])      # vlan-1672 vlan-1672

blk.attributes["from"] = "vlan-1700"                         # writes through, like att does

The same applies to the other two accessors, whose keys are chosen by APIC rather than by you: set.metadata["Content-Type"] for a response header, set.annotations["polled at"] for an annotation you named with a space. getattr(tree.att, "from") works too, but the dictionary reads better. And set.imdata returns the untouched imdata list of the response.

Navigating the tree

children returns the direct children of a tree; search() walks the entire response at any depth. Both return a ListOfAciTree, empty if nothing matched.

tree = api.lookup_by_dn("uni/tn-infra/ap-access/epg-default", rsp_subtree="full")

for child in tree.children:
    print(child.att.rn)                     # rscustQosPol, rsbd, ...

search() filters on the ACI class, on attribute values, or on both. Omit the class to match every object in the response:

tree = api.lookup_by_dn("uni/tn-infra", rsp_subtree="full")

tree.search("fvBD")                         # every bridge domain, at any depth
tree.search("fvBD", arpFlood="yes")         # ... with arpFlood set to yes
tree.search(unicastRoute="no")              # any class, filtered on an attribute
tree.search()                               # everything in the response

direct_children=True keeps the search shallow: the objects of the set and their direct children are looked at, nothing below them. It is children with a class and an attribute filter on top.

tenant = api.lookup_by_dn("uni/tn-infra", rsp_subtree="full")

tenant.search("fvAEPg")                     # every EPG of the tenant, however deep
tenant.search("fvAp", direct_children=True) # the application profiles, not what is inside them
tenant.search("fvAEPg", direct_children=True)   # nothing: an EPG hangs under an application profile

Objects nested in an rsp-subtree response are returned by APIC with an rn but no dn. YaPyAci computes the missing dn from the parent while walking, so att.dn is always available on whatever children or search() hands back.

Set-level information

ltree = api.lookup_by_class("fvTenant")

ltree.total_count                           # totalCount reported by APIC
len(ltree)                                  # number of trees actually returned
bool(ltree)                                 # False when the query matched nothing
ltree.metadata["Content-Length"]            # HTTP response headers of the query
ltree.subscription_id                       # '' unless the query was a subscription

total_count and len() differ when the response is paginated: total_count is what APIC says exists, len() is what came back in this page.

Annotations

Annotations are your own key/values, carried along with a set. They are useful to tag where a result came from when several APICs or several queries are merged in one place. They survive copies and show up in the JSON representation of the object.

tree = api.lookup_by_dn("uni/tn-common")
tree.annot.fabric = "brussels"
tree.annot.polled_at = "2020-08-31T10:00:00"

print(tree.annot.fabric)                    # brussels
print(tree.annotations)                     # {'fabric': 'brussels', 'polled_at': '...'}

Converting between sets and trees

from yapyaci import to_tree, to_ltree

to_tree(ltree)                              # ListOfAciTree -> AciTree
to_ltree(tree)                              # AciTree -> ListOfAciTree

ltree.tree()                                # same thing, as methods
tree.ltree()

to_tree() raises CastingFailure if the set holds more than one tree. Indexing is often easier: ltree[0] is already an AciTree.

Printing

str() renders as JSON on one line: an AciTree as its object, a ListOfAciTree as the list of all of them. Annotations come along if the set carries any.

repr() of an AciTree is the same JSON, but a ListOfAciTree summarises itself instead — a set can hold a whole fabric, and dumping it into a terminal is rarely what you wanted:

ltree = api.lookup_by_class("fvTenant")

ltree                                       # <ListOfAciTree 8 object(s): fvTenant>
print(ltree)                                # [{"fvTenant": {"attributes": {...}}}, ...]

The summary names up to three of the ACI classes in the set, and says how many objects the query matched when APIC paginated its answer — <ListOfAciTree 2 of 200 object(s): fvTenant>.

mpprint() pretty-prints the first tree of a set, so call it on an AciTree, or loop over a ListOfAciTree:

from yapyaci import mpprint

mpprint(api.lookup_by_dn("uni/tn-common", rsp_subtree="full"))

Modifying a tree

pop() removes an attribute and update_cls() rewrites the ACI class. Both change the tree in place and return None — useful to strip APIC-generated attributes before comparing or storing a configuration.

tree.pop("modTs")
tree.pop("uid")

Note that children and search() also modify the underlying response in place, by filling in the dn of nested objects. Use copy.deepcopy() if you need an untouched copy; AciSet, AciTree and ListOfAciTree all support copy and deepcopy.

Building sets from a file

ObjectSetBuilder turns a JSON payload saved on disk into the same objects a lookup returns, without talking to an APIC. Record a response once, then replay it in tests and offline tooling:

from yapyaci import ObjectSetBuilder, load_ltree, load_tree

# The full builder, when you want the metadata, the annotations or several casts
builder = ObjectSetBuilder("tenants.json", annotations={"source": "lab fabric"})
tenants = builder.build()                       # ListOfAciTree
first = builder.build()[0]                      # AciTree

# Or the one-liners
tenants = load_ltree("tenants.json")            # ListOfAciTree
tenant = load_tree("tenant.json")               # AciTree, raises CastingFailure if the file holds several

Three shapes of file are accepted, so a fixture can be trimmed down to the interesting part:

In the file	Becomes
{"totalCount": "2", "imdata": [ ... ]}	The set as APIC returned it, subscriptionId included
[ {"fvTenant": ...}, {"fvTenant": ...} ]	A set over that list
{"fvTenant": {"attributes": ...}}	A set of one object

The result is an ordinary AciSet, so search(), children, att and the rest behave exactly as they do on a query result. build_set() returns the raw AciSet when a subscription id matters, and ObjectSetBuilder.from_string() / from_payload() build from a string or an already decoded payload instead of a file. Anything that is not readable, not valid JSON, or not shaped like an APIC payload raises InvalidJsonPayload.

# A test running against a recorded fabric state instead of a live one
tenant = load_tree("recorded_tenant.json")

assert [bd.att.name for bd in tenant.search("fvBD")] == ["main.alpha.bd"]

Subscribing to events

Rather than polling, you can have APIC push changes to you. This is the part of the library that only EApi has — see Api or EApi? for the rest of what it adds.

Subscribing takes a WebSocket for the events to arrive on, and a subscription that APIC will keep feeding as long as it is refreshed. YaPyAci opens the first and refreshes the second in background threads; you supply a handler for the events.

Note that EApiCtx takes keyword arguments only.

import time
from yapyaci import EApi, EApiCtx, PrintStream

ctx = EApiCtx(ip="192.168.1.1", username="my_user", password="my_password")
api = EApi(ctx)

stream = api.start_stream(PrintStream)      # open the WebSocket first
api.subscribe(cls="eventRecord")
api.subscribe(cls="faultInfo")
api.subscribe(cls="healthInst")

time.sleep(20)
api.stop_stream()

The order matters: subscribe() raises SubscriptionFailure if no stream is open for the context, because there would be nowhere for the events to arrive. start_stream() waits up to 5 seconds for the WebSocket to come up, raises WsOpenFailure if it does not or if the context already has one, and otherwise returns the stream thread — join() on it to keep the main thread alive and notice when the stream dies:

stream = api.start_stream(PrintStream)
api.subscribe(cls="faultInfo")
stream.join()                               # returns when the WebSocket closes

subscribe() takes the same arguments as lookup() and returns its result, so the current state of what you subscribed to is available without a second query. Subscription ids are refreshed every 180 seconds for as long as the stream is open; if a refresh fails the stream is torn down rather than left receiving a subset of the events.

Handling events yourself

PrintStream pretty-prints whatever arrives, which is mostly useful to see that a subscription works. To do something with the events, subclass StreamHandler and implement _handler(). It is called with the raw JSON message for every event APIC pushes:

import json
from yapyaci import EApi, EApiCtx, StreamHandler, ListOfAciTree

class FaultLogger(StreamHandler):
    def _handler(self, msg):
        for tree in ListOfAciTree(json.loads(msg)):
            print("{0}: {1} - {2}".format(tree.cls, tree.att.dn, tree.att.descr))

ctx = EApiCtx(ip="192.168.1.1", username="my_user", password="my_password")
api = EApi(ctx)

stream = api.start_stream(FaultLogger)
api.subscribe(cls="faultInfo")
stream.join()

StreamHandler is a threading.Thread, so anything you keep on the instance is shared with your handler. Extra arguments given to start_stream() are passed on to the handler's constructor.

Exceptions

All exceptions live in yapyaci.exception, and all of them descend from YaPyAciError, so anything this library raises can be caught in one place:

from yapyaci import YaPyAciError
from yapyaci.exception import AuthFailure, ConnFailure, QueryFailure

try:
    api.lookup_by_class("fvTenant")
except YaPyAciError as error:                # anything below, and nothing else
    ...

Between the two sit four families, to catch a kind of failure without listing its members: TransportError (the cluster could not be reached), AuthenticationError (the session could not be established or kept), QueryError (a query was rejected, or its answer could not be read) and StreamError (the WebSocket or a subscription failed).

Exception	Raised when
AuthFailure	The credentials were rejected by APIC.
AuthRefreshFailure	A token refresh was rejected; the session is closed and the next query logs in again.
ConnFailure	No member of the cluster could serve the request. The message lists what each one answered.
QueryFailure	APIC answered a query with something other than HTTP 200.
InvalidLookupRequest	lookup() was called with neither a dn nor a cls.
CastingFailure	A set holding several trees was cast to a single AciTree.
InvalidJsonPayload	A member answered something it called JSON that does not parse, or ObjectSetBuilder was given a file or a payload it could not turn into a set.
WsOpenFailure	The WebSocket did not open within 5 seconds, or the context already has one.
SubRefreshFailure	A subscription could not be refreshed.
SubscriptionFailure	subscribe() was called with no stream open.
InvalidCtx	An EApi was handed a plain ApiCtx, which holds none of the state it needs.

An HTTP 403 on a query clears the token on the context, so the next query authenticates again; the query that got the 403 still raises QueryFailure and is not retried for you.

Logging

YaPyAci logs through the standard logging package and never configures it — that is left to the application. Each module logs under its own name (yapyaci.api, yapyaci.eapi, yapyaci.object_set, yapyaci.stream_handler), so they can be tuned separately.

import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# The query path is chatty; keep it for troubleshooting only
logging.getLogger('yapyaci.api').setLevel(logging.WARNING)

INFO reports authentication, stream and subscription lifecycle; WARNING reports poisoned cluster members and failed refreshes; DEBUG reports every request and its response size.

Notes and limitations

• Queries only. Every method reads from APIC. Creating, modifying or deleting managed objects is not supported — the only POST the module sends is the login.
• TLS certificates are not verified. Both the HTTPS connections and the WebSocket accept whatever certificate APIC presents, which is what makes the default self-signed certificate of a fresh fabric work out of the box. Keep that in mind on an untrusted network.
• Results share their underlying data. An AciTree taken from a ListOfAciTree points at the same dictionaries as the set it came from, and the tree-walking helpers fill in missing dns in place. copy.deepcopy() when you need isolation.
• Responses are read as JSON. The XML form of the APIC API is not supported.

License

MIT. See LICENSE.