anac

Python Async NetBox API Client, based on httpx and pydantic

Documentation

https://timeforplanb123.github.io/anac

Features

• Minimalistic interface
• Async only
• Python interpreter autocompletion
• Supports NetBox 2.x, 3.x
• Flexibility. All the objects are coroutines or coroutine iterators
• Simple integration with parsers (TextFSM, TTP)

Quick Start

Install

Please, at first, check the dependencies in pyproject.toml and create new virtual environment if necessary and then:

with pip:

pip install anac 

with git:

git clone https://github.com/timeforplanb123/anac.git
cd anac 
pip install .
# or
poetry install

Simple Examples

Api Instantiating

from anac import api

a = api(
    "https://demo.netbox.dev",
    token="cf1dc7b04de5f27cfc93aba9e3f537d2ad6fdf8c",
)
# get openapi spec and create attributes/endpoints   
await a.openapi()

get some device and patch it

In [1]: some_device = await a.dcim_devices(get={"name": "dmi01-rochster-sw01"})

In [2]: some_device.name
Out[2]: 'dmi01-rochster-sw01'

In [3]: some_device.device_type
Out[3]:
{'id': 7,
 'url': 'https://demo.netbox.dev/api/dcim/device-types/7/',
 'display': 'C9200-48P',
 'manufacturer': {'id': 3,
  'url': 'https://demo.netbox.dev/api/dcim/manufacturers/3/',
  'display': 'Cisco',
  'name': 'Cisco',
  'slug': 'cisco'},
 'model': 'C9200-48P',
 'slug': 'c9200-48p'}

In [4]: some_device.status
Out[4]: {'value': 'active', 'label': 'Active'}

In [5]: some_device = await some_device(patch={"status": "failed"})

In [6]: some_device.status
Out[6]: {'value': 'failed', 'label': 'Failed'}

get some 2 devices and put + patch them

In [7]: some_devices = await a.dcim_devices(
    ...:     get=[{"name": "dmi01-rochster-sw01"}, {"name": "dmi01-rochester-rtr01"}]
    ...: )

# EndpointAsIterator is a coroutine iterator with 2 coroutines
In [8]: some_devices
Out[8]: EndpointAsIterator(api=Api, url='https://demo.netbox.dev/api', endpoint='/dcim/devices/')

In [9]: import asyncio

# run 2 coroutines in the event loop
In [10]: some_devices = await asyncio.gather(*some_devices)

# EndpointId is a NetBox '/dcim/devices/' object and coroutine
In [11]: some_devices
Out[11]: 
[EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/21/', endpoint='/dcim/devices/'),
 EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/8/', endpoint='/dcim/devices/')]

In [12]: patch_some_devices = [coro(patch={"status": "failed"}) for coro in some_devices]

# run 2 coroutines in the event loop
In [13]: patch_some_devices = await asyncio.gather(*patch_some_devices)

# EndpointId is a coroutine, again
In [14]: patch_some_devices
Out[14]: 
[EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/21/', endpoint='/dcim/devices/{id}/'),
 EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/8/', endpoint='/dcim/devices/{id}/')]

In [15]: patch_some_devices[0].name
Out[15]: 'dmi01-rochster-sw01'

In [16]: patch_some_devices[0].status
Out[16]: {'value': 'failed', 'label': 'Failed'}

In [17]: patch_some_devices[1].name
Out[17]: 'dmi01-rochester-rtr01'

In [18]: patch_some_devices[1].status
Out[18]: {'value': 'failed', 'label': 'Failed'}

get all devices

In [19]: all_devices = await a.dcim_devices(get={})
# or
In [20]: all_devices = await a.dcim_devices()

# EndpointIdIterator is an coroutine iterator with EndpointId objects
In [21]: all_devices
Out[21]: EndpointIdIterator(api=Api, url='https://demo.netbox.dev/api', endpoint='/dcim/devices/')

In [22]: len(all_devices)
Out[22]: 50

In [23]: all_devices[49]
Out[23]: EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/95/', endpoint='/dcim/devices/')

In [24]: all_devices[49].name
Out[24]: 'ncsu118-distswitch1'

# by default, 'limit' parameter = 50, but you can run 'get' request with custom 'limit'
In [25]: all_devices = await a.dcim_devices(get={"limit": 100})

In [26]: len(all_devices)
Out[26]: 75

In [27]: all_devices[74]
Out[27]: EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/106/', endpoint='/dcim/devices/')

In [28]: all_devices[74].id
Out[28]: 106

get all devices and post 2 new devices

In [29]: all_test = await a.dcim_devices(
    ...:     get={},
    ...:     post=[
    ...:         {
    ...:             "name": "test1",
    ...:             "device_role": 1,
    ...:             "site": 1,
    ...:             "device_type": 1,
    ...:             "status": "planned",
    ...:         },
    ...:         {
    ...:             "name": "test2",
    ...:             "device_role": 1,
    ...:             "site": 1,
    ...:             "device_type": 1,
    ...:             "status": "planned",
    ...:         },
    ...:     ],
    ...: )

# run 3 coroutines in the event loop
In [30]: all_test = await asyncio.gather(*all_test)

# EndpointIdIterator is an coroutine iterator with EndpointId objects
# EndpointId is a NetBox '/dcim/devices/' object and coroutine
In [31]: all_test
Out[31]: 
[EndpointIdIterator(api=Api, url='https://demo.netbox.dev/api', endpoint='/dcim/devices/'),
 EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/110/', endpoint='/dcim/devices/'),
 EndpointId(api=Api, url='https://demo.netbox.dev/api/dcim/devices/111/', endpoint='/dcim/devices/')]

In [32]: all_test[0][49].name
Out[32]: 'ncsu118-distswitch1'

In [33]: all_test[1].name
Out[34]: 'test1'

In [35]: all_test[2].name
Out[35]: 'test2'

# httpx.Response is available with .response attribute
In [36]: all_test[1].response.json()
Out[36]:
{'id': 110,
 'url': 'https://demo.netbox.dev/api/dcim/devices/110/',
 'display': 'test1',
 'name': 'test1',
 'device_type': {'id': 1,
  'url': 'https://demo.netbox.dev/api/dcim/device-types/1/',
  'display': 'MX480',
  'manufacturer': {'id': 7,
  ...
}