Walking operations¶

Walk a table¶

Walk one subtree with SNMP GETNEXT using the following options:

with SNMPv2c, community ‘public’
over IPv4/UDP
to an Agent at demo.pysnmp.com:161
for all objects under IF-MIB::ifTable
stopping at the end of that subtree rather than running on into the next one
based on asyncio I/O framework

walk_cmd() reissues GETNEXT until the subtree runs out. next_cmd() is the single request underneath it, and being lexicographic over the whole MIB it walks straight past the end of whatever was asked for – which is what makes hand-written walks report rows the caller never wanted.

Functionally similar to:

$ snmpwalk -v2c -c public demo.pysnmp.com IF-MIB::ifTable

import asyncio

from pysnmp.hlapi.asyncio import *


async def run():
    snmpEngine = SnmpEngine()

    async for errorIndication, errorStatus, errorIndex, varBinds in walk_cmd(
        snmpEngine,
        CommunityData("public"),
        UdpTransportTarget(("demo.pysnmp.com", 161)),
        ContextData(),
        ObjectType(ObjectIdentity("IF-MIB", "ifTable")),
    ):
        if errorIndication:
            print(errorIndication)
            break

        if errorStatus:
            print(
                "{} at {}".format(
                    errorStatus.prettyPrint(),
                    errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
                )
            )
            break

        for varBind in varBinds:
            print(" = ".join([x.prettyPrint() for x in varBind]))

    snmpEngine.closeDispatcher()


asyncio.run(run())

Download script.

Bulk walk a table¶

Walk one subtree with SNMP GETBULK using the following options:

with SNMPv2c, community ‘public’
over IPv4/UDP
to an Agent at demo.pysnmp.com:161
for all objects under IF-MIB::ifTable
25 rows per request
stopping at the end of that subtree, with the overshoot trimmed
based on asyncio I/O framework

An agent answering GETBULK fills maxRepetitions rows whether or not they belong to the subtree asked for, so the last response of a table walk routinely runs past the end of it. bulk_walk_cmd() trims that tail; bulk_cmd(), being the single request underneath, hands it back.

Being an async generator, this can also be stopped early – leave the loop and no further request is sent.

Functionally similar to:

$ snmpbulkwalk -v2c -c public -Cn0 -Cr25 demo.pysnmp.com IF-MIB::ifTable

import asyncio

from pysnmp.hlapi.asyncio import *


async def run():
    snmpEngine = SnmpEngine()

    async for errorIndication, errorStatus, errorIndex, varBinds in bulk_walk_cmd(
        snmpEngine,
        CommunityData("public"),
        UdpTransportTarget(("demo.pysnmp.com", 161)),
        ContextData(),
        0,
        25,
        ObjectType(ObjectIdentity("IF-MIB", "ifTable")),
    ):
        if errorIndication:
            print(errorIndication)
            break

        if errorStatus:
            print(
                "{} at {}".format(
                    errorStatus.prettyPrint(),
                    errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
                )
            )
            break

        for varBind in varBinds:
            print(" = ".join([x.prettyPrint() for x in varBind]))

    snmpEngine.closeDispatcher()


asyncio.run(run())

Download script.

Bulk walk MIB¶

Send a series of SNMP GETBULK requests using the following options:

with SNMPv3, user ‘usr-none-none’, no authentication, no privacy
over IPv4/UDP
to an Agent at localhost:161
for all OIDs past SNMPv2-MIB::system
run till end-of-mib condition is reported by Agent
based on asyncio I/O framework

Functionally similar to:

$ snmpbulkwalk -v3 -lnoAuthNoPriv -u public -Cn0 -Cr50

localhost SNMPv2-MIB::system

import asyncio

from pysnmp.hlapi.asyncio import *


async def run(varBinds):
    snmpEngine = SnmpEngine()
    while True:
        errorIndication, errorStatus, errorIndex, varBindTable = await bulk_cmd(
            snmpEngine,
            CommunityData("public"),
            UdpTransportTarget(("localhost", 161)),
            ContextData(),
            0,
            50,
            *varBinds,
        )
        if errorIndication:
            print(errorIndication)
            break
        elif errorStatus:
            print(
                "{} at {}".format(
                    errorStatus.prettyPrint(),
                    errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
                )
            )
        else:
            for varBindRow in varBindTable:
                for varBind in varBindRow:
                    print(" = ".join([x.prettyPrint() for x in varBind]))

        varBinds = varBindTable[-1]
        if is_end_of_mib(varBinds):
            break


asyncio.run(
    run([ObjectType(ObjectIdentity("TCP-MIB")), ObjectType(ObjectIdentity("IP-MIB"))])
)

Download script.