pysnmp-sync-adapter 1.0.2.post1

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().

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())

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

Contributing

Contributions are welcome! Please follow standard guidelines:

• Fork the repository
• Create a feature branch
• Submit a Pull Request

License

EUPL-1.2 License - See LICENSE for details.