Niwaki

https://github.com/k3l0-dev/niwaki
Admin
Typed, design-first Python SDK for Cisco ACI (APIC): declare tenants, BDs, EPGs and contracts as...
Tokens:33,931
Snippets:431
Trust Score:7.5
License:Apache-2.0
Update:2 months ago
Show doc for...
Tokens:
Raw
================
LIBRARY RULES
================
- 1. Niwaki is the typed, design-first layer over the Cisco APIC REST API — what an ORM is to SQL. Express configuration through the design DSL (`niwaki.design`); never hand-build `uni/...` JSON payloads or URL strings.
- 2. Do not configure through the `Niwaki`/`AsyncNiwaki` facade — it has no write methods (it navigates, reads, queries, deletes). All configuration writes go through the design DSL.
- 3. Canonical flow: `tenant("prod").vrf("main")` then `tenant("prod").bd("web", arp_flooding=True).bind(vrf="main")` then `.push(aci)` — designs are built detached (no session, no I/O) and applied afterwards.
- 4. Push modes: `push(aci)` is one atomic POST (all-or-nothing, default); `push(aci, mode="plan")` is the dry-run diff (`.creates/.updates/.unchanged`) — use it before any change.
- 5. When partial progress matters, `push(aci, mode="staged")` applies ordered waves and raises `StagedPushError` naming the failing DNs.
- 6. A design never removes what it does not declare — there is no reconciliation or pruning. Deletion is an explicit act: `aci.tenant("prod").bd("old").delete()`.
- 7. Use readable field names in models and the DSL (`arp_flooding=True`); wire names are legal in exactly two places: query filters (`where(arpFlood=True)` goes to the APIC verbatim) and raw payloads/item access (`mo["arpFlood"]`).
- 8. `bool(query)` raises by design — use `.exists()`. `.one()` demands exactly one match (`NoResultError`/`MultipleResultsError`); use `.first()` when zero matches is acceptable.
- 9. `with_faults()` embeds fault children in results; it does not filter. To restrict results to faulted objects, use `only_faulted()`.
- 10. Query any of the ~15,450 ACI classes by wire-name string — `aci.query("topSystem")` — including operational classes with no generated model; results still expose readable field names via the read catalogue.
- 11. Class and property discovery is offline via `niwaki.catalog` (`search`, `describe`, `prop_meta`, `generated_classes`) — no APIC connection needed. `catalog.generated_classes()` enumerates the ~2,200 classes with typed models.
- 12. Subscribe to live changes with `aci.query(fvBD).under(dn).subscribe()`; iterate typed CREATED/MODIFIED/DELETED events. Stats classes can never push — `StatsClassNotSubscribableError` fires before any network call.
- 13. A subscription GAP event means the socket reconnected and events were lost forever (the APIC has no replay) — reconcile with a fresh read when it matters.
- 14. Imports like `from niwaki.models.fv.fvBD import fvBD` are valid at runtime (module alias); static analyzers may flag them — expected, do not fix. If tooling insists, the statically-resolvable path is `niwaki.models._generated.fv.fvBD`.
- 15. Do not emit APIs that do not exist: the facade has no `create/push/update/bind`, models have no `add_*`/`bind_*` builders, and filter operators `contains`/`isdigit` do not exist (use `wcard`).
- 16. Niwaki is not cobra or acitoolkit — never import `cobra.*` or mix its `ConfigRequest` pattern with niwaki code.
- 17. Catch typed exceptions: every SDK error subclasses `NiwakiError` (auth/transport/API/query/design/subscription branches). `catalog` lookups raise `UnknownClassError`, which is also a `KeyError`.
- 18. Everything has an async twin with the same names: `AsyncNiwaki`, `await query.fetch()`, `async for event in sub` — sync and async mirror each other.
- 19. Requires Python 3.12+; models track the APIC 6.0 schema release. Install: `pip install niwaki` (or `uv add niwaki`).
- 20. This context covers the niwaki SDK only; full documentation lives at https://k3l0-dev.github.io/niwaki/.

================
CODE SNIPPETS
================
### Setup development environment

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/installation.md

Clone the repository and sync dependencies for SDK development.

```bash
git clone https://github.com/k3l0-dev/niwaki
cd niwaki
uv sync --extra dev
```

--------------------------------

### Verify installation

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/installation.md

Check that the package is correctly installed by printing the version.

```bash
python -c "import niwaki; print(niwaki.__version__)"
```

--------------------------------

### Build and Open Documentation

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Installs documentation dependencies and generates the static HTML site.

```bash
uv sync --extra docs
bash scripts/docs.sh open     # build + open docs/_build/html/index.html
```

--------------------------------

### Sync Development Environment

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Installs project dependencies required for development.

```bash
uv sync --extra dev
```

--------------------------------

### Initialize Niwaki connection

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/push-modes.md

Setup the design configuration and establish a connection to the APIC.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("prod").vrf("main")
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
```

--------------------------------

### Install niwaki from offline wheelhouse

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/installation.md

Install the package in restricted environments without reaching out to PyPI.

```bash
unzip niwaki-<version>-offline-wheelhouse.zip -d wheelhouse
pip install --no-index --find-links=wheelhouse niwaki
```

```bash
uv pip install --no-index --find-links=wheelhouse niwaki
```

--------------------------------

### Open a session

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/comparison.md

Establishes a connection to the APIC. The cobra example requires manual session management, while niwaki handles connection lifecycle.

```python
from cobra.mit.access import MoDirectory
from cobra.mit.session import LoginSession

ls = LoginSession("https://apic.example.com", "admin", "secret")
moDir = MoDirectory(ls)
moDir.login()
# … your code …
moDir.logout()
```

```python
from niwaki import Niwaki

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
```

--------------------------------

### Define Baseline Configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/day-2-changes.md

Initial setup of a bridge domain with ARP flooding disabled.

```python
from niwaki import Niwaki
from niwaki.design import tenant

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

baseline = tenant("commerce")
baseline.vrf("prod")
baseline.bd("bd-db", unicast_routing=True).bind(vrf="prod").subnet("10.30.30.1/24")
baseline.push(aci)
```

--------------------------------

### Define CI pipeline jobs

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/gitops-pipeline.md

Example YAML configuration for CI jobs to handle planning on merge requests and applying on main branch merges.

```yaml
plan:            # merge-request job
  script: python apply.py --plan

apply:           # main-branch job, behind a protected environment
  script: python apply.py
```

--------------------------------

### Describe, apply, and observe configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/quickstart.md

Demonstrates building a detached design, performing a dry-run plan, pushing the configuration, and reading back the state.

```python
from niwaki.design import tenant

config = tenant("prod", description="my first tenant")
config.vrf("main")
config.bd("web", unicast_routing=True).bind(vrf="main").subnet("10.0.1.1/24")

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    plan = config.push(aci, mode="plan")  # dry run — nothing written
    print(plan.creates)  # every DN that would be created

    config.push(aci)  # one atomic POST

    bd = aci.tenant("prod").bd("web").read()  # observe it back
    assert bd.unicast_routing is True
```

--------------------------------

### Connect and push configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/observing.md

Initializes a tenant configuration and pushes it to the APIC using a Niwaki client.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("prod")
config.bd("web", unicast_routing=True).bind(vrf="main")
config.vrf("main")

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
config.push(aci)
```

--------------------------------

### Apply a configuration design

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/design-first.md

Demonstrates how to define an infrastructure policy and push it to the controller using the Niwaki client.

```python
from niwaki import Niwaki
from niwaki.design import infra

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    infra().cdp_policy("cdp-on", admin_state="disabled").push(aci)
```

--------------------------------

### Instantiate and use a model class

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/reference/api/models.md

Demonstrates importing a model, initializing it with parameters, and accessing its properties or wire payload.

```python
from niwaki.models.fv.fvBD import fvBD

bd = fvBD(name="web", unicast_routing=True)  # validated at construction
bd.rn  # "BD-web"
bd.to_apic()  # wire payload, ACI attribute names
```

--------------------------------

### Initialize and use AsyncNiwaki

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/async.md

Demonstrates basic usage of the asynchronous client to push designs and read objects.

```python
import asyncio

from niwaki import AsyncNiwaki
from niwaki.design import tenant

config = tenant("prod")
config.bd("web", unicast_routing=True).bind(vrf="main")
config.vrf("main")


async def apply() -> None:
    async with AsyncNiwaki("https://apic.example.com", "admin", "secret") as aci:
        await config.push(aci)  # designs are transport-agnostic
        bd = await aci.tenant("prod").bd("web").read()
        assert bd.unicast_routing is True


asyncio.run(apply())
```

--------------------------------

### Perform a day-2 configuration update

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/design-dsl.md

Use set() to define specific field changes and push() with the plan mode to preview drifts before applying.

```python
patch = tenant("prod").bd("backend").set(description="patched")
patch.push(aci, mode="plan")  # exactly one field change reported
patch.push(aci)
```

--------------------------------

### Define and compile a design configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/inside-the-dsl.md

Demonstrates creating a tenant configuration, adding bridge domains and VRFs, and compiling it into an APIC payload.

```python
from niwaki.design import tenant

config = tenant("shop")
config.bd("web", arp_flooding=True).bind(vrf="prod")
config.vrf("prod")

payload = config.to_payload()

bd = payload["polUni"]["children"][0]["fvTenant"]["children"][0]["fvBD"]
assert bd["attributes"] == {"name": "web", "arpFlood": "true"}
assert bd["children"] == [{"fvRsCtx": {"attributes": {"tnFvCtxName": "prod"}}}]
```

--------------------------------

### Construct a client using a context manager

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/connection.md

Use the context manager to automatically handle authentication on entry and session closure on exit.

```python
from niwaki import Niwaki

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

--------------------------------

### Verifying convergence with plan

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/testing.md

Use the plan mode to verify that a pushed configuration results in no further changes.

```python
config = build_shop("acme")
config.push(aci)

assert config.push(aci, mode="plan").has_changes is False
```

--------------------------------

### Plan the deployment

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/onboard-tenant.md

Compare the design against the current fabric state to identify required changes without applying them.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print(f"{len(plan.creates)} objects to create")
assert plan.has_changes is True
```

--------------------------------

### Provisioning a tenant with Niwaki

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/migrate-from-cobra.md

Simplified tenant provisioning using Niwaki's detached design patterns and automatic schema validation.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("commerce")
config.vrf("prod")
config.bd("bd-web").bind(vrf="prod")

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    config.push(aci)
```

--------------------------------

### Plan the configuration push

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/application-contracts.md

Connects to the APIC and generates a plan to preview changes before applying them.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print(f"{len(plan.creates)} objects to create")
```

--------------------------------

### Construct a client manually

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/connection.md

Use connect() and close() for long-lived services or interactive sessions where a context manager is unsuitable.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
tenants = aci.query("fvTenant").fetch()
aci.close()
```

--------------------------------

### Configure credentials via environment variables

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/connection.md

Omit constructor arguments to fall back to environment variables, which is recommended for CI/CD and containerized environments.

```python
import os

os.environ["APIC_HOST"] = "https://apic.example.com"
os.environ["APIC_USERNAME"] = "admin"
os.environ["APIC_PASSWORD"] = "secret"

with Niwaki() as aci:  # everything comes from the environment
    ...
```

--------------------------------

### Plan and push configuration to APIC

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/microsegmentation-esg.md

Connects to the APIC and pushes the defined configuration, using a plan mode to preview changes.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print(f"{len(plan.creates)} objects to create")
config.push(aci)
```

--------------------------------

### Declare and configure objects

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Shows how to declare an object and configure its attributes simultaneously using maker keyword arguments.

```python
tn = tenant("prod")
tn.bd("web", arp_flooding=True, unicast_routing=False)  # declare + configure
```

--------------------------------

### Navigate nodes by vocabulary

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/observing.md

Demonstrates navigating the object tree using DN-scoped handles and reading typed instances.

```python
bd = aci.tenant("prod").bd("web")  # NiwakiNode at uni/tn-prod/BD-web
assert bd.dn == "uni/tn-prod/BD-web"
mo = bd.read()  # typed fvBD instance
assert mo.unicast_routing is True  # human-readable field names
```

--------------------------------

### Execute push modes

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/push-modes.md

Demonstrates the three primary push modes: strict, staged, and plan.

```python
report = config.push(aci)  # strict (default)
report = config.push(aci, mode="staged")
plan = config.push(aci, mode="plan")
```

--------------------------------

### Declarative Provisioning with Design DSL

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Build a detached design tree and push it to the APIC controller.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = (
    tenant("prod")
    .app("shop")
        .epg("frontend").bind(bd="frontend").consume("fe-to-be")
        .epg("backend").bind(bd="backend").provide("fe-to-be")
    .bd("frontend")
        .set(unicast_routing=True)
        .bind(vrf="prod")
        .subnet("10.0.1.1/24")
    .bd("backend")
        .set(unicast_routing=True)
        .bind(vrf="prod")
        .subnet("10.0.2.1/24")
    .vrf("prod")
    .filter("api")
        .entry("rest", tcp=8080)
    .contract("fe-to-be")
        .set(scope="vrf")
        .subject("api").bind(filter="api")
)

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    config.push(aci, mode="strict")
```

--------------------------------

### Deploy test configuration to fabric

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/fabric-audit.md

Initializes a connection and pushes a tenant configuration to the APIC for subsequent auditing.

```python
from niwaki import Niwaki
from niwaki.design import tenant

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

config = tenant("commerce")
config.vrf("prod")
for name, gw in {"bd-web": "10.30.10.1/24", "bd-app": "10.30.20.1/24"}.items():
    config.bd(name, unicast_routing=True).bind(vrf="prod").subnet(gw)
config.app("storefront").epg("web").bind(bd="bd-web")
config.push(aci)
```

--------------------------------

### Provisioning a tenant with Cobra

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/migrate-from-cobra.md

Standard tenant provisioning using the official Cobra SDK, requiring explicit session management and request objects.

```python
from cobra.mit.access import MoDirectory
from cobra.mit.session import LoginSession
from cobra.mit.request import ConfigRequest
from cobra.model.fv import Tenant, Ctx, BD, RsCtx

ls = LoginSession("https://apic.example.com", "admin", "secret")
moDir = MoDirectory(ls)
moDir.login()

uniMo = moDir.lookupByDn("uni")
tenantMo = Tenant(uniMo, "commerce")
Ctx(tenantMo, "prod")
bdMo = BD(tenantMo, "bd-web")
RsCtx(bdMo, tnFvCtxName="prod")

req = ConfigRequest()
req.addMo(tenantMo)
moDir.commit(req)
moDir.logout()
```

--------------------------------

### Importing the Niwaki Catalog

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/discovery.md

Initializes the offline catalogue module.

```python
from niwaki import catalog
```

--------------------------------

### Implement the CI runner apply logic

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/gitops-pipeline.md

A function to handle configuration application, supporting a plan-only mode to preview changes before execution. Credentials should be sourced from environment variables.

```python
import os

from niwaki import Niwaki


def apply(config, *, plan_only: bool) -> bool:
    with Niwaki() as aci:  # APIC_* environment variables
        plan = config.push(aci, mode="plan")
        for dn in plan.creates:
            print(f"+ {dn}")
        for dn, fields in plan.updates.items():
            for field, (current, desired) in fields.items():
                print(f"~ {dn} {field}: {current!r} -> {desired!r}")
        if plan_only or not plan.has_changes:
            return plan.has_changes
        config.push(aci)
        return plan.has_changes


# In CI these come from the runner's secret store; set here only so the page is
# self-contained and runnable.
os.environ["APIC_HOST"] = "https://apic.example.com"
os.environ["APIC_USERNAME"] = "admin"
os.environ["APIC_PASSWORD"] = "from-the-secret-store"

changed = apply(build(), plan_only=True)  # the merge-request job
assert changed is True  # empty fabric: the tenant is new
```

--------------------------------

### Define and Push ACI Configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/index.md

Use the design DSL to define tenants, VRFs, and bridge domains, then apply the configuration to an APIC instance using the Niwaki client.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("prod", description="my first tenant")
config.vrf("main")
config.bd("web", unicast_routing=True).bind(vrf="main").subnet("10.0.1.1/24")

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    config.push(aci, mode="plan")  # dry run — see the diff first
    config.push(aci)  # one atomic POST
```

--------------------------------

### Apply and Verify Changes

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/day-2-changes.md

Push the changes to the APIC and confirm the state has converged.

```python
change.push(aci)

bd = aci.tenant("commerce").bd("bd-db").read()
assert bd.arp_flooding is True

assert change.push(aci, mode="plan").has_changes is False
```

--------------------------------

### Multi-domain Design Provisioning

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Configure fabric and access policies in a single atomic operation.

```python
from niwaki.design import design

config = design()
config.fabric().datetime_policy("prod-ntp").ntp_provider("10.0.0.1")
inf = config.infra()
inf.vlan_pool("prod", "static").range("vlan-100", "vlan-199")
config.phys_dom("prod-phys").bind(vlan_pool="prod")
inf.aaep("prod-aaep").bind(domain="prod-phys")
config.tenant("prod").app("shop").epg("web").bind_dn(domain="uni/phys-prod-phys")

config.push(aci)          # everything above in ONE atomic POST
```

--------------------------------

### Plan and Push Configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/sr-mpls-handoff.md

Connects to the APIC and executes the plan and push operations for both infra and tenant configurations.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

infra_plan = inf.push(aci, mode="plan")
print(f"infra: {len(infra_plan.creates)} objects")
inf.push(aci)

tenant_plan = t.push(aci, mode="plan")
print(f"tenant: {len(tenant_plan.creates)} objects")
t.push(aci)
```

--------------------------------

### Push the configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/application-contracts.md

Applies the defined configuration to the APIC.

```python
report = config.push(aci)
assert report.request_count == 1
```

--------------------------------

### Verify the configuration

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/application-contracts.md

Queries the APIC to confirm that EPGs and contracts were created as expected.

```python
epgs = aci.tenant("commerce").query("fvAEPg").fetch()
assert {e.name for e in epgs} == {"web", "app", "db"}

provided = aci.tenant("commerce").query("fvRsProv").fetch()
assert {p.name for p in provided} == {"web-to-app", "app-to-db"}

assert config.push(aci, mode="plan").has_changes is False
```

--------------------------------

### Subscribe to a live stream

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/subscribing.md

Use Query.subscribe() to receive a synchronous snapshot followed by a continuous stream of events.

```python
from niwaki.models.fv.fvBD import fvBD

with aci.query(fvBD).under("uni/tn-prod").subscribe() as sub:
    for bd in sub.initial:  # the synchronous snapshot, first
        print("already there:", bd.dn)
    for event in sub:  # then the live stream, forever
        print(event.kind, event.dn)
```

--------------------------------

### Implement check-before-write pattern

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/migrate-from-cobra.md

Use the plan mode to review changes before executing the final push to the APIC.

```python
change = tenant("commerce").bd("bd-web").set(arp_flooding=True)

plan = change.push(aci, mode="plan")
print(plan.updates or plan.creates)  # review artifact — nothing written yet

change.push(aci)
assert change.push(aci, mode="plan").has_changes is False
```

--------------------------------

### Initialize a cursor

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Demonstrates how a maker returns a cursor for the declared child object.

```python
from niwaki.design import tenant

tn = tenant("prod")  # a cursor at the tenant
bd = tn.bd("web")  # a cursor at the new BD — a different position
assert bd.dn == "uni/tn-prod/BD-web"
```

--------------------------------

### Contract wiring comparison

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/comparison.md

Demonstrates the reduction in boilerplate code when defining application profiles, EPGs, and contracts.

```python
from cobra.model.fv import Ap, AEPg, RsBd, RsProv, RsCons
from cobra.model.vz import Filter, Entry, BrCP, Subj, RsSubjFiltAtt

apMo = Ap(tnMo, "shop")
webMo = AEPg(apMo, "web")
RsBd(webMo, tnFvBDName="web")
RsProv(webMo, tnVzBrCPName="web-api")
filterMo = Filter(tnMo, "http")
Entry(filterMo, "e1", etherT="ip", prot="tcp", dFromPort="8080", dToPort="8080")
brcpMo = BrCP(tnMo, "web-api")
subjMo = Subj(brcpMo, "s1")
RsSubjFiltAtt(subjMo, tnVzFilterName="http")
```

```python
epg = config.app("shop").epg("web")
epg.bind(bd="web").provide("web-api")
config.filter("http").entry("e1", tcp=8080)
config.contract("web-api").subject("s1").bind(filter="http")
config.push(aci)
```

--------------------------------

### Define the tenant design

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/onboard-tenant.md

Create an in-memory configuration for a tenant, its VRF, and associated bridge domains with subnets.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("commerce", description="Retail commerce platform")
config.vrf("prod")

segments = {
    "bd-web": "10.30.10.1/24",  # public-facing web tier
    "bd-app": "10.30.20.1/24",  # application tier
    "bd-db": "10.30.30.1/24",  # database tier
}
for name, gateway in segments.items():
    config.bd(name, unicast_routing=True).bind(vrf="prod").subnet(gateway)
```

--------------------------------

### Build and push a design asynchronously

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/async-at-scale.md

Demonstrates using AsyncNiwaki to push a design and perform a read operation within an async context.

```python
import asyncio

from niwaki import AsyncNiwaki
from niwaki.design import tenant


def build(name: str) -> object:
    config = tenant(name)
    config.vrf("prod")
    config.bd("bd-web", unicast_routing=True).bind(vrf="prod").subnet("10.30.10.1/24")
    return config


async def onboard_one() -> None:
    async with AsyncNiwaki("https://apic.example.com", "admin", "secret") as aci:
        await build("commerce").push(aci)
        bd = await aci.tenant("commerce").bd("bd-web").read()
        assert bd.unicast_routing is True


asyncio.run(onboard_one())
```

--------------------------------

### Use smart keyword filters

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/observing.md

Demonstrates how value types influence APIC operators and how to inspect compiled filters using .build().

```python
from niwaki.query import any_of, anybit

# list -> membership OR
_, params = aci.query(fvBD).where(name=["web", "db"]).build()
assert params["query-target-filter"] == 'or(eq(fvBD.name,"web"),eq(fvBD.name,"db"))'

# a "*" makes a wildcard
_, params = aci.query(fvBD).where(name="prod-*").build()
assert params["query-target-filter"] == 'wcard(fvBD.name,"prod-*")'

# a set stays a bitmask equality (Flags fields), never a membership OR
_, params = aci.query("fvSubnet").where(scope={"public", "shared"}).build()
assert params["query-target-filter"] == 'eq(fvSubnet.scope,"public,shared")'

# explicit helpers: any_of(...), and anybit(...) to match a single bit of a mask
faults = aci.query("faultInst").where(code=any_of("F0467", "F1394")).fetch()
open_subnets = aci.query("fvSubnet").where(anybit("fvSubnet.scope", "shared")).fetch()
```

--------------------------------

### Perform typed queries with Niwaki

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Demonstrates reading data using vocabulary navigation and the query builder for filtering and scoping.

```python
from niwaki import Niwaki
from niwaki.models.fv.fvBD import fvBD

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    # Vocabulary navigation, no class imports needed
    bd = aci.tenant("prod").bd("frontend").read()

    # Query builder: filters, scoping, enrichment, pagination
    # (filters address the APIC attribute names — the wire side)
    bds = aci.query(fvBD).where(arpFlood=True).under("uni/tn-prod").fetch()
    n = aci.tenant("prod").query(fvBD).count()

    # Any of the ~15k APIC classes by name (read-only/operational included)
    nodes = aci.query("topSystem").naming_only().fetch()
```

--------------------------------

### Reference objects with bind()

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Demonstrates using bind() to create relationships between objects, which are resolved at push time.

```python
tn = tenant("prod")
tn.bd("web").bind(vrf="main")  # the fvRsCtx relation, resolved to the VRF below
tn.vrf("main")  # declared after — closed world, not ordering
```

--------------------------------

### Perform a day-2 patch

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Shows how to use set() to perform a targeted update on a specific field.

```python
patch = tenant("prod").bd("web").set(description="patched")
# push(mode="plan") reports exactly one field change; parents touch nothing.
```

--------------------------------

### Verify the deployment

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/onboard-tenant.md

Read back the fabric state to confirm the configuration and ensure no further changes are required.

```python
bd = aci.tenant("commerce").bd("bd-web").read()
assert bd.unicast_routing is True

bds = aci.tenant("commerce").query("fvBD").fetch()
assert {b.name for b in bds} == {"bd-web", "bd-app", "bd-db"}

assert config.push(aci, mode="plan").has_changes is False
```

--------------------------------

### Define and Push Design

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/design-dsl.md

Constructs a multi-domain configuration tree and pushes it to the APIC in a single atomic operation.

```python
from niwaki.design import design

config = design()
config.fabric().datetime_policy("prod-ntp").ntp_provider("10.0.0.1")
config.infra().vlan_pool("prod", "static").range("vlan-100", "vlan-199")
config.tenant("prod").vrf("main")
config.push(aci)  # the three domains in ONE atomic POST
```

--------------------------------

### Configure TLS with a custom CA bundle

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/connection.md

Provide a path to a PEM bundle to verify connections against a private or enterprise CA.

```python
with Niwaki(
    "https://apic.example.com",
    "admin",
    "secret",
    verify_ssl="/etc/ssl/certs/corp-ca.pem",  # PEM bundle, loaded eagerly
) as aci:
    ...
```

--------------------------------

### Push configuration in staged mode

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/turn-up-a-rack.md

Connects to the APIC and pushes configuration using plan and staged modes to manage object creation.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print(f"{len(plan.creates)} objects to create")

report = config.push(aci, mode="staged")
```

--------------------------------

### Configure attributes with set()

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Illustrates using set() to configure attributes on an object that has already been declared.

```python
tn = tenant("prod")
bd = tn.bd("web")
bd.set(arp_flooding=True)
bd.set(unicast_routing=False)  # merges — both attributes are now set
```

--------------------------------

### Access model field documentation

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/reference/api/models.md

Shows how to retrieve field descriptions directly from the model class metadata within an IDE.

```python
fvBD.model_fields["arp_flooding"].description
```

--------------------------------

### Push Configuration to APIC

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/external-connectivity.md

Connects to the APIC and pushes the design configuration, optionally using a plan mode to preview changes.

```python
aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print(f"{len(plan.creates)} objects")
config.push(aci)
```

--------------------------------

### Initialize design cursors

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/design-dsl.md

Define initial design cursors for tenants, applications, and infrastructure components to be used for reference binding.

```python
config = design()
tn = config.tenant("prod")
epg = tn.app("shop").epg("web")
vrf = tn.vrf("main")
inf = config.infra()
aaep = inf.aaep("prod-aaep")
port_selector = inf.access_port_profile("leaf101").port_selector("esxi", "range")
```

--------------------------------

### Minimal watcher for object changes

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/watch-for-changes.md

Opens a live push stream for a specific object class and prints events as they occur. Requires an active Niwaki session and a defined query scope.

```python
from niwaki import Niwaki
from niwaki.models.fv.fvBD import fvBD
from niwaki.query import EventKind

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    with aci.query(fvBD).under("uni/tn-commerce").subscribe() as sub:
        print(f"watching {len(sub.initial)} existing BD(s)")
        for event in sub:
            match event.kind:
                case EventKind.CREATED:
                    print(f"+ {event.dn}")
                case EventKind.MODIFIED:
                    print(f"~ {event.dn} changed: {sorted(event.mo.model_fields_set)}")
                case EventKind.DELETED:
                    print(f"- {event.dn}")
                case EventKind.GAP:
                    print("! reconnected — events during the gap were not replayed")
                case EventKind.REFRESH_FAILED:
                    print("! a scheduled refresh was rejected (informational)")
```

--------------------------------

### Apply design with push()

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/cursors.md

Use push() to perform I/O and apply the design to the fabric. The mode parameter determines whether the operation is a plan, staged, or strict update.

```python
from niwaki import Niwaki

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

config = tenant("prod")
config.bd("web").bind(vrf="main")
config.vrf("main")
report = config.push(aci, mode="plan")  # what would change, no write
```

--------------------------------

### Plan Changes

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/troubleshooting.md

Preview the impact of a push operation without applying changes to the fabric.

```python
from niwaki import Niwaki

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")

plan = config.push(aci, mode="plan")
print("creates  :", plan.creates)
print("updates  :", plan.updates)
print("converged:", not plan.has_changes)
```

--------------------------------

### Fan out operations with gather()

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/async-at-scale.md

Uses gather() to execute multiple push operations concurrently for batch provisioning.

```python
async def onboard_batch(names: list[str]) -> set[str]:
    async with AsyncNiwaki("https://apic.example.com", "admin", "secret") as aci:
        await aci.gather(*(build(name).push(aci) for name in names))
        tenants = await aci.query("fvTenant").fetch()
        return {t.name for t in tenants}


provisioned = asyncio.run(onboard_batch(["shop-eu", "shop-us", "shop-apac"]))
assert {"shop-eu", "shop-us", "shop-apac"} <= provisioned
```

--------------------------------

### Define rack infrastructure design

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/turn-up-a-rack.md

Initializes a multi-domain design and configures interface policies, VLAN pools, and physical domains for a new rack.

```python
from niwaki import Niwaki
from niwaki.design import design

config = design()
inf = config.infra()

# Interface policies — reusable, so named after behaviour, not location
inf.cdp_policy("cdp-on", admin_state="enabled")
inf.lldp_policy("lldp-on", receive_state="enabled", transmit_state="enabled")
inf.lacp_policy("lacp-active", mode="active")

# Encap: pool -> physical domain -> AAEP
inf.vlan_pool("commerce-static", "static").range("vlan-1410", "vlan-1449")
config.phys_dom("commerce-phys").bind(vlan_pool="commerce-static")
inf.aaep("commerce-aaep").bind(domain="commerce-phys")

# The vPC policy group ties the interface policies and the AAEP together
groups = inf.func_profile()
groups.port_channel("esxi-vpc", link_aggregation_type="node").bind(
    aaep="commerce-aaep", cdp="cdp-on", lldp="lldp-on", lacp="lacp-active"
)

# Interface profile: which ports...
ports = inf.access_port_profile("rack-b7-ports")
uplinks = ports.port_selector("esxi-uplinks", "range")
uplinks.port_block("blk1", from_port_id=10, to_port_id=11)
uplinks.bind(policy_group="esxi-vpc")

# ...on which switches
leaves = inf.leaf_profile("rack-b7-leaves")
leaves.leaf_selector("pair", "range").node_block("blk1", from_node_id=101, to_node_id=102)
leaves.bind(interface_profile="rack-b7-ports")

# Fabric side: the explicit vPC protection pair
pair = config.fabric().vpc_protection().vpc_pair("101-102", logical_pair_id="7")
pair.node("101")
pair.node("102")
```

--------------------------------

### Connect to APIC

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/quickstart.md

Establishes a connection to the APIC using a context manager for automatic session handling.

```python
from niwaki import Niwaki

with Niwaki("https://apic.example.com", "admin", "secret") as aci:
    ...
```

--------------------------------

### Querying ACI classes with Cobra

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/why.md

Demonstrates the standard Cobra approach using raw string filters and wire-format attribute names.

```python
moDir.lookupByClass("fvTenant", propFilter='and(eq(fvTenant.name, "Tenant1"))')
```

--------------------------------

### Connecting to an APIC

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/testing.md

Establish a connection to the APIC for integration testing.

```python
from niwaki import Niwaki

aci = Niwaki.connect("https://apic.example.com", "admin", "secret")
```

--------------------------------

### Perform asynchronous operations with AsyncNiwaki

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Shows how to use the asynchronous mirror of the Niwaki API to perform concurrent reads and push configurations.

```python
from niwaki import AsyncNiwaki
from niwaki.models.fv.fvTenant import fvTenant

async with AsyncNiwaki("https://apic.example.com", "admin", "secret") as aci:
    tenants, bd = await aci.gather(
        aci.query(fvTenant).fetch(),
        aci.tenant("prod").bd("frontend").read(),
    )
    await config.push(aci, mode="strict")   # the design DSL is async-ready too
```

--------------------------------

### Day-2 Configuration Changes

Source: https://github.com/k3l0-dev/niwaki/blob/main/README.md

Perform attribute-less upserts for specific configuration changes.

```python
from niwaki.design import infra

flip = infra().cdp_policy("cdp-on", admin_state="disabled")
flip.push(aci, mode="plan")   # shows exactly one field change
flip.push(aci)
```

--------------------------------

### Configure ACI using Cobra

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/why.md

Imperative configuration using the Cobra SDK, requiring explicit knowledge of class names, relations, and parent-child plumbing.

```python
from cobra.model.fv import Tenant, Ctx, BD, RsCtx, Ap, AEPg, RsBd
from cobra.mit.request import ConfigRequest

fvTenantMo = Tenant(uniMo, "ExampleCorp")
Ctx(fvTenantMo, "private-net1")
fvBDMo = BD(fvTenantMo, "bridge-domain1")
RsCtx(fvBDMo, tnFvCtxName="private-net1")
fvApMo = Ap(fvTenantMo, "WebApp")
fvAEPgMo = AEPg(fvApMo, "WebEPG")
RsBd(fvAEPgMo, tnFvBDName="bridge-domain1")

configReq = ConfigRequest()
configReq.addMo(fvTenantMo)
moDir.commit(configReq)
```

--------------------------------

### Define typed parameters in the DSL

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/design-first.md

Demonstrates how the DSL translates typed parameters into specific APIC object structures.

```python
entry("rest", tcp=8080)
```

--------------------------------

### Filtered query comparison

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/comparison.md

Shows how to perform filtered object lookups without manual string-based filter construction.

```python
bds = moDir.lookupByClass("fvBD", propFilter='and(eq(fvBD.arpFlood, "no"))')
```

```python
bds = aci.query("fvBD").where(arpFlood="no").fetch()
```

--------------------------------

### Iterate and fetch query results

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/observing.md

Shows lazy iteration, slicing, and methods for retrieving single objects or checking existence.

```python
# Iterate lazily — every result reads the same, generated class or not:
# .dn and obj["wireName"] work uniformly across all ~15,000 APIC classes.
for bd in aci.query(fvBD):
    print(bd.dn, bd["name"])

# A leading slice caps the result (server-side page size), lazily:
first_page = list(aci.query(fvBD)[:50])

# Exactly one object, or a typed error:
web = aci.query(fvBD).where(name="web").one()
assert web["name"] == "web"

# Cheap existence check:
assert aci.query(fvBD).where(name="web").exists()
```

--------------------------------

### Configure static path for EPG

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/turn-up-a-rack.md

Defines a static path for an EPG to a vPC pair using a literal DN for the physical topology.

```python
tn = config.tenant("commerce")
tn.vrf("prod")
tn.bd("bd-web", unicast_routing=True).bind(vrf="prod").subnet("10.30.10.1/24")

web = tn.app("storefront").epg("web").bind(bd="bd-web").bind(domain="commerce-phys")
web.static_path(
    "topology/pod-1/protpaths-101-102/pathep-[esxi-vpc]",
    encap="vlan-1410",
    deployment_immediacy="immediate",
)
```

--------------------------------

### Execute idempotent apply operations

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/gitops-pipeline.md

Demonstrates running the apply function in non-plan mode to commit changes and verifying idempotence on subsequent runs.

```python
apply(build(), plan_only=False)  # main-branch job: writes
assert apply(build(), plan_only=False) is False  # re-run converges to a no-op
```

--------------------------------

### Configure SR-MPLS Underlay Infrastructure

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/sr-mpls-handoff.md

Defines the global MPLS configuration, QoS policies, and L3Out within the infra tenant.

```python
from niwaki import Niwaki
from niwaki.design import tenant

# Border leaves that carry the handoff — an SR-MPLS infra L3Out rides border
# *leaves* (the APIC rejects spine nodes on an MPLS L3Out).
border_leaves = [("border-101", 101), ("border-102", 102)]

inf = tenant("infra")

# The MPLS global label policy is the fabric singleton "default".  It cannot be
# modified — it is referenced attribute-free, purely to resolve the binding.
inf.mpls_global_configuration("default")
inf.mpls_interface_policy("mpls-backbone", description="MPLS handoff interface policy.")
inf.bgp_peer_prefix_policy("sr-mpls-limit", max_number_of_prefixes=20000, max_prefix_action="log")

# MPLS EXP marking is supported only under tenant infra, so it lives here and
# the node profile binds it by name below.
qos = inf.mpls_custom_qos_policy(
    "mpls-exp-marking", description="MPLS EXP marking for the handoff."
)
qos.mpls_ingress_rule("0", "3", prio="level3", target="CS3", target_cos="3", description="EXP in.")
qos.mpls_egress_rule("0", "31", target_cos="5", target_exp="5", description="DSCP out.")

# Encap plumbing for the handoff links
inf.infra().vlan_pool("sr-mpls-underlay", "static").range(
    "vlan-2690", "vlan-2699", allocation_mode="static", role="external"
)
inf.l3_dom("sr-mpls-idom").bind(vlan_pool="sr-mpls-underlay")

# Reference the fabric infra VRF (overlay-1) — attribute-free, so it is only made
# resolvable, never reconfigured.
inf.vrf("overlay-1")

# The infra SR-MPLS L3Out, its MPLS-external config, and the provider label
out = (
    inf.l3out("sr-mpls-infra", mpls_enabled=True).bind(vrf="overlay-1").bind(domain="sr-mpls-idom")
)
out.mpls_external(description="MPLS handoff config.").bind(mpls_global_configuration="default")
out.provider_label("sr-backbone", tag="green", description="SR-MPLS provider label.")
```

--------------------------------

### Bind access group with interface protocols

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/turn-up-a-rack.md

Configures a bare-metal access group with specific discovery protocols.

```python
groups.access_group("bare-metal").bind(aaep=..., cdp=..., lldp=...)
```

--------------------------------

### Define ESGs and contracts with Niwaki

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/microsegmentation-esg.md

Configures ESGs for web, app, and database tiers using tag selectors and binds them to a production VRF.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("commerce")
config.vrf("prod")

app = config.app("storefront")

web = app.esg("esg-web").bind(vrf="prod")
web.tag_selector("tier", "web")
web.consume("web-to-app")

svc = app.esg("esg-app").bind(vrf="prod")
svc.tag_selector("tier", "app")
svc.provide("web-to-app")
svc.consume("app-to-db")

db = app.esg("esg-db").bind(vrf="prod")
db.tag_selector("tier", "db")
db.provide("app-to-db")

config.filter("f-http").entry("http", tcp=8080)
config.filter("f-postgres").entry("pg", tcp=5432)
config.contract("web-to-app").set(scope="vrf").subject("http").bind(filter="f-http")
config.contract("app-to-db").set(scope="vrf").subject("sql").bind(filter="f-postgres")
```

--------------------------------

### Create a BD with a subnet in a VRF

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/comparison.md

Configures a Bridge Domain, subnet, and VRF association. Niwaki abstracts parent lookups and relation classes, providing built-in design validation.

```python
from cobra.model.fv import Tenant, Ctx, BD, RsCtx, Subnet
from cobra.mit.request import ConfigRequest

uniMo = moDir.lookupByDn("uni")
tnMo = Tenant(uniMo, "acme")
Ctx(tnMo, "prod")
bdMo = BD(tnMo, "web")
Subnet(bdMo, "10.30.1.1/24")
RsCtx(bdMo, tnFvCtxName="prod")

req = ConfigRequest()
req.addMo(tnMo)
moDir.commit(req)
```

```python
from niwaki.design import tenant

config = tenant("acme")
config.vrf("prod")
config.bd("web", unicast_routing=True).bind(vrf="prod").subnet("10.30.1.1/24")
config.push(aci)
```

--------------------------------

### Access schema documentation

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/design-dsl.md

Retrieve Cisco's class definitions directly from the APIC schemas via the object's docstring.

```python
from niwaki.design import tenant

doc = type(tenant("acme")).bd.__doc__
assert "unique layer 2 forwarding domain" in doc  # Cisco's own definition
```

--------------------------------

### Plan and Validate Changes

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/day-2-changes.md

Generate a plan to verify the specific field updates before applying them to the fabric.

```python
plan = change.push(aci, mode="plan")

dn = "uni/tn-commerce/BD-bd-db"
assert plan.updates[dn] == {"arp_flooding": (False, True)}
assert plan.creates == []
```

--------------------------------

### Verify Environment Variables for Authentication

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/troubleshooting-connection.md

Check which environment variables are set to debug LoginError issues caused by empty or missing credentials.

```python
import os

seen = {k: bool(os.environ.get(k)) for k in ("APIC_HOST", "APIC_USERNAME", "APIC_PASSWORD")}
print(seen)
```

--------------------------------

### Define tenant, VRF, bridge domains, and EPGs

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/cookbook/application-contracts.md

Configures the tenant structure and defines EPGs with their respective provide/consume relationships.

```python
from niwaki import Niwaki
from niwaki.design import tenant

config = tenant("commerce")
config.vrf("prod")
config.bd("bd-web", unicast_routing=True).bind(vrf="prod").subnet("10.30.10.1/24")
config.bd("bd-app", unicast_routing=True).bind(vrf="prod").subnet("10.30.20.1/24")
config.bd("bd-db", unicast_routing=True).bind(vrf="prod").subnet("10.30.30.1/24")

app = config.app("storefront")
app.epg("web").bind(bd="bd-web").consume("web-to-app")
app.epg("app").bind(bd="bd-app").provide("web-to-app").consume("app-to-db")
app.epg("db").bind(bd="bd-db").provide("app-to-db")
```

--------------------------------

### Query.subscribe()

Source: https://github.com/k3l0-dev/niwaki/blob/main/docs/guide/subscribing.md

Turns a query into a live push stream for a specific class, providing an initial snapshot and subsequent event updates.

```APIDOC
## Query.subscribe()

### Description
Converts a query into a live push stream. The client receives notifications when a matching object is created, modified, or deleted.

### Usage
```python
with aci.query(ModelClass).under("scope").subscribe() as sub:
    for item in sub.initial: # Initial snapshot
        print(item)
    for event in sub: # Live stream
        print(event.kind, event.dn)
```

### SubscriptionEvent Properties
- **kind** (EventKind) - The type of event (CREATED, MODIFIED, DELETED, GAP, REFRESH_FAILED).
- **mo** (Object) - The typed object associated with the event (not present for GAP or REFRESH_FAILED).
- **dn** (string) - The distinguished name of the object.
- **subscription_ids** (list) - Identifiers for the subscription.

### EventKind Constants
- **CREATED** - Object was created.
- **MODIFIED** - Object was modified.
- **DELETED** - Object was deleted.
- **GAP** - Subscription reconnected; events during the gap are lost.
- **REFRESH_FAILED** - A scheduled refresh was rejected.
```