pysnmp-sync-adapter 1.0.3

Synchronous wrapper adapters for pysnmp asyncio HLAPI

pysnmp-sync-adapter

Lightweight Synchronous Adapter for PySNMP AsyncIO HLAPI

---

This package provides lightweight, blocking wrappers around pysnmp.hlapi.v1arch.asyncio and pysnmp.hlapi.v3arch.asyncio, enabling synchronous use of the SNMPv1 high-level API without requiring direct asyncio management.

Features

• Drop-in synchronous alternatives to PySNMP's async-HLAPI: get_cmd_sync, next_cmd_sync, set_cmd_sync, bulk_cmd_sync, walk_cmd_sync, bulk_walk_cmd_sync.
• Supports both v1arch and v3arch PySNMP architectures, automatically selected or configurable via the PYSNMP_ARCH environment variable.
• Supports both IPv4 and IPv6 transport targets via UdpTransportTarget and Udp6TransportTarget.
• Reuses or creates the default shared event loop (asyncio.get_event_loop()), ensuring integration efficiency.
• Sync wrappers accept an optional timeout parameter (in seconds) that limits the total execution time using asyncio.wait_for().
• Minimizes connection overhead by reusing pre-created transport instances when calling create_transport().
• In addition, through pysnmp_sync_adapter.legacy_wrappers, it supports the old etingof/pysnmp v5 HLAPI

These adapters allow to call the familiar HLAPI functions in a purely synchronous style (e.g. in scripts, GUIs like Tkinter, or blocking contexts) without having to manage asyncio directly.

This restores the synchronous experience familiar from earlier PySNMP versions. Native sync HLAPI wrappers were deprecated in recent releases in favor of asyncio.

Provided Methods

Synchronous Function	AsyncIO Equivalent
get_cmd_sync	get_cmd
next_cmd_sync	next_cmd
set_cmd_sync	set_cmd
bulk_cmd_sync	bulk_cmd
walk_cmd_sync	walk_cmd (async-gen)
bulk_walk_cmd_sync	bulk_walk_cmd (async-gen)

Internal Utilities

• ensure_loop() — Retrieves the current default event loop via asyncio.get_event_loop(), or creates and sets one if none exists. Ensures one loop is available per thread.
• create_transport() — Synchronously awaits the create() factory method of UdpTransportTarget or Udp6TransportTarget, returning a ready-to-use transport object.
• _sync_coro() — Executes a coroutine to completion on the shared event loop, with optional timeout support via asyncio.wait_for(). Handles already-running loops by scheduling a future.
• _sync_agen() — Collects all items from an async generator (e.g., walk_cmd) into a list by internally awaiting it with _sync_coro().
• make_sync() — Higher-order function that wraps PySNMP async-HLAPI coroutines into synchronous functions, propagating optional timeout arguments.

By avoiding per-call event loop instantiation and by reusing transport targets, this implementation significantly reduces runtime overhead in tight polling or query loops.

---

Installation

pip install pysnmp-sync-adapter

Quick Start

from pysnmp.hlapi.v1arch.asyncio import *
from pysnmp_sync_adapter import get_cmd_sync, create_transport

err, status, index, var_binds = get_cmd_sync(
    SnmpDispatcher(),
    CommunityData('public', mpModel=0),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))
)

for name, val in var_binds:
    print(f'{name} = {val}')

Usage

To ensure compatibility with the selected PySNMP architecture (v1arch or v3arch), make sure to import pysnmp.hlapi.v3arch.asyncio (or v1arch) before importing from pysnmp_sync_adapter. For example:

from pysnmp.hlapi.v3arch.asyncio import *  # Must come first (or v1arch)

from pysnmp_sync_adapter import (
    get_cmd_sync,
    next_cmd_sync,
    set_cmd_sync,
    bulk_cmd_sync,
    walk_cmd_sync,
    bulk_walk_cmd_sync,
    create_transport
)

This ensures that the adapter binds to the appropriate internal PySNMP modules. If omitted or imported in the wrong order, pysnmp_sync_adapter may fallback to v1arch even when v3arch is desired.

Alternatively, the environment variable PYSNMP_ARCH can be set to "v3arch" (or "v1arch"). Example:

import os
os.environ["PYSNMP_ARCH"] = "v3arch"  # or "v1arch"

from pysnmp_sync_adapter import get_cmd_sync  # etc.

This method is particularly useful in larger applications or testing scenarios where import order might be harder to control.

High-level v1arch sync

import asyncio
import platform
from pysnmp.hlapi.v1arch.asyncio import *
from pysnmp_sync_adapter import (
    get_cmd_sync,
    next_cmd_sync,
    set_cmd_sync,
    bulk_cmd_sync,
    walk_cmd_sync,
    bulk_walk_cmd_sync,
    create_transport
)


if platform.system() == "Windows":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
community = "public"
dispatcher = SnmpDispatcher()
auth_data = CommunityData(community, mpModel=0)

print("\n--> get_cmd_sync")
error_indication, error_status, error_index, var_binds = get_cmd_sync(
    dispatcher,
    auth_data,
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> set_cmd_sync")
error_indication, error_status, error_index, var_binds = set_cmd_sync(
    dispatcher,
    auth_data,
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0), "Linux i386"),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> next_cmd_sync")
error_indication, error_status, error_index, var_binds = next_cmd_sync(
    dispatcher,
    auth_data,
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "system")),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> bulk_cmd_sync")
error_indication, error_status, error_index, var_binds = bulk_cmd_sync(
    dispatcher,
    CommunityData("public"),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    0,
    2,
    ObjectType(ObjectIdentity("SNMPv2-MIB", "system")),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> walk_cmd_sync")
objects = walk_cmd_sync(
    dispatcher,
    auth_data,
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr")),
    timeout=30  # Notice that this optional timeout is added to the adapter
)
for error_indication, error_status, error_index, var_binds in objects:
    for name, val in var_binds:
        print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> bulk_walk_cmd_sync")
objects = bulk_walk_cmd_sync(
    dispatcher,
    CommunityData("public"),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    0,
    25,
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr")),
    timeout=30  # Notice that this optional timeout is added to the adapter
)
for error_indication, error_status, error_index, var_binds in objects:
    for name, val in var_binds:
        print(name.prettyPrint(), "=", val.prettyPrint())

High-level v3arch sync

import asyncio
import platform
from pysnmp.hlapi.v3arch.asyncio import *
from pysnmp_sync_adapter import (
    get_cmd_sync,
    next_cmd_sync,
    set_cmd_sync,
    bulk_cmd_sync,
    walk_cmd_sync,
    bulk_walk_cmd_sync,
    create_transport
)

if platform.system() == "Windows":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
community = "public"
engine = SnmpEngine()

print("\n--> get_cmd_sync")
error_indication, error_status, error_index, var_binds = get_cmd_sync(
    engine,
    CommunityData(community),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> set_cmd_sync")
error_indication, error_status, error_index, var_binds = set_cmd_sync(
    engine,
    CommunityData(community),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0), "Linux i386"),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> next_cmd_sync")
error_indication, error_status, error_index, var_binds = next_cmd_sync(
    engine,
    CommunityData(community),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "system")),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> bulk_cmd_sync")
error_indication, error_status, error_index, var_binds = bulk_cmd_sync(
    engine,
    CommunityData("public"),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    0,
    2,
    ObjectType(ObjectIdentity("SNMPv2-MIB", "system")),
)
print(error_indication, error_status, error_index)
for name, val in var_binds:
    print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> walk_cmd_sync")
objects = walk_cmd_sync(
    engine,
    CommunityData(community),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr")),
    timeout=30  # Notice that this optional timeout is added to the adapter
)
for error_indication, error_status, error_index, var_binds in objects:
    for name, val in var_binds:
        print(name.prettyPrint(), "=", val.prettyPrint())

print("\n--> bulk_walk_cmd_sync")
objects = bulk_walk_cmd_sync(
    engine,
    CommunityData("public"),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    0,
    25,
    ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr")),
    timeout=30  # Notice that this optional timeout is added to the adapter
)
for error_indication, error_status, error_index, var_binds in objects:
    for name, val in var_binds:
        print(name.prettyPrint(), "=", val.prettyPrint())

---

Support of legacy etingof/pysnmp v5 HLAPI

This adapter provides compatibility for code written against the legacy etingof/pysnmp v5 API (https://github.com/etingof/pysnmp) using synchronous SNMP commands.

Supported synchronous functions

The adapter implements wrappers for:

• getCmd
• setCmd
• nextCmd
• bulkCmd
• walkCmd
• bulkWalkCmd
• UdpTransportTarget

These wrappers preserve the iterator-based usage of pysnmp.hlapi but operate using blocking, synchronous calls underneath.

Example Usage

from pysnmp.hlapi.v1arch.asyncio import *
from pyasn1.type.univ import OctetString as OctetStringType
from pysnmp_sync_adapter.legacy_wrappers import UdpTransportTarget, getCmd

timeout = 2
retries = 2
iterator = getCmd(
    SnmpDispatcher(),
    CommunityData('public', mpModel=0),
    UdpTransportTarget(
        ("demo.pysnmp.com", 161),
        timeout,  # optional parameter
        retries  # optional parameter
    ),
    ('1.3.6.1.2.1.1.1.0', None)
)

for response in iterator:
    errorIndication, errorStatus, errorIndex, varBinds = response
    if errorIndication:
        print(errorIndication)
    elif errorStatus:
        print(errorStatus)
    else:
        for varBind in varBinds:
            print(' = '.join([x.prettyPrint() for x in varBind]))

Notes

UdpTransportTarget

The adapter supports two legacy initialization forms:

UdpTransportTarget(("host", port), timeout, retries)
UdpTransportTarget(("host", port, timeout, retries))

Both forms correctly map to the underlying transport constructor, omitting timeout and retries if None.

errorIndication

If the error indication errorIndication is present, the returned message is a string.

Besides, if the message includes "before timeout", it will be augmented with " - timed out" for compatibility matching.

Module Reference: pysnmp_sync_adapter.legacy_wrappers

Function	Description
getCmd(...)	Yields a single (errInd, errStat, errIdx, varBinds) from get_cmd_sync
setCmd(...)	Same as getCmd but for set_cmd_sync
nextCmd(...)	Uses next_cmd_sync
bulkCmd(...)	Uses bulk_cmd_sync
walkCmd(...)	Uses walk_cmd_sync
bulkWalkCmd(...)	Uses bulk_walk_cmd_sync
UdpTransportTarget(...)	Legacy-compatible wrapper

This extension allows legacy SNMPv1/v2c/v3 code to run unchanged while taking advantage of simplified synchronous operation. Ideal for code needing backward compatibility.

---

Notes and limitations

• These adapters block the calling thread until the SNMP operation completes.
• They rely on the default asyncio event loop obtained via asyncio.get_event_loop(). If no loop is set, one is created and registered. They do not create isolated loops.
• Since PySNMP uses the default event loop bound to the current thread, invoking these synchronous wrappers from a thread that is already running an event loop may cause deadlocks or RuntimeError. To use them safely in such environments, run them from a separate thread.
• A timeout (in seconds) can be optionally passed to all sync wrappers; it limits the total wall-clock time of the SNMP operation using asyncio.wait_for(). On timeout, asyncio.TimeoutError is raised.
• The underlying socket layer’s timeouts (e.g. UdpTransportTarget(..., timeout=2)) still apply, and should be set appropriately to avoid low-level blocking.
• These wrappers do not forcibly cancel low-level socket operations. A timeout interrupts the coroutine, but not the transport at the OS level.

This repository uses the public SNMP simulation service at demo.pysnmp.com, provided courtesy of Lextudio. Please ensure network access to demo.pysnmp.com:161 is available when running the tests (python -m pytest).

Reference documentation

• PySNMP: https://docs.lextudio.com/snmp/

• PySNMP API Reference: https://docs.lextudio.com/pysnmp/v7.1/docs/api-reference

License

EUPL-1.2 License - See LICENSE for details.