Synchronous wrapper adapters for pysnmp asyncio HLAPI

These details have not been verified by PyPI

Project links

Project description

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. It supports a variety of legacy interfaces.

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 Python SNMP library v5.0.24 HLAPI
- the etingof/pysnmp v5 HLAPI
Also, through pysnmp_sync_adapter.cmdgen_wrappers, it supports code written against the cmdgen SNMP library appearing in pysnmp.entity.rfc3413.oneliner.

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

Using v1arch:

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}')

Using v3arch:

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

err, status, index, var_binds = get_cmd_sync(
    SnmpEngine(),
    CommunityData('public', mpModel=0),
    create_transport(UdpTransportTarget, ("demo.pysnmp.com", 161), timeout=2),
    ContextData(),
    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())

Supporting other libraries

This adapter provides compatibility for code written against other libraries using synchronous SNMP commands. It allows legacy SNMPv1/v2c/v3 code to run unchanged for backward compatibility, while taking advantage of simplified synchronous operation.

Implements 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`
`Udp6TransportTarget(...)`	Legacy-compatible wrapper
`UdpTransportTarget(...)`	Legacy-compatible wrapper

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

Support of the Python SNMP library v5.0.24 HLAPI

Compatibility layer for code written against the Python SNMP library v5 HLAPI (https://github.com/pysnmp/pysnmp) using synchronous SNMP commands.

This library uses SnmpEngine() and ContextData(). It requires legacy_wrappers with v3arch.

Example Usage

from pysnmp.hlapi.v3arch.asyncio import *
from pysnmp_sync_adapter.legacy_wrappers import UdpTransportTarget, getCmd

for errorIndication, errorStatus, errorIndex, varBinds in getCmd(
    SnmpEngine(),
    CommunityData('public', mpModel=0),
    UdpTransportTarget(("demo.pysnmp.com", 161)),
    ContextData(),
    ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))
):
    if errorIndication:
        print(errorIndication, errorStatus, errorIndex, varBinds)
    elif errorStatus:
        print(errorIndication, errorStatus, errorIndex, varBinds)
    else:
        for name, val in varBinds:
            print(name, "=", val)

Support of legacy `etingof/pysnmp` v5 HLAPI

Compatibility layer for code written against the legacy etingof/pysnmp v5 HLAPI (https://github.com/etingof/pysnmp) using synchronous SNMP commands.

This library uses SnmpDispatcher() and does not use ContextData(). It requires legacy_wrappers with v1arch.

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, errorStatus, errorIndex, varBinds)
    elif errorStatus:
        print(errorIndication, errorStatus, errorIndex, varBinds)
    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.

Support of the cmdgen SNMP library apearing in pysnmp oneliner

Compatibility layer for code written against the cmdgen SNMP library appearing in pysnmp.entity.rfc3413.oneliner, using synchronous SNMP commands.

It needs v3arch, transparently inserts the required SnmpEngine() and ContextData() parameters for SNMPv3 (v3arch.asyncio) calls, and wraps OID tuples in ObjectType(ObjectIdentity(...)). Legacy UDP and UDP6 transports, including timeout and retry arguments, are preserved.

Example Usage

from pysnmp.hlapi.v3arch.asyncio import *
import pysnmp_sync_adapter.cmdgen_wrappers as cmdgen
cmd_gen = cmdgen.CommandGenerator()
transport = cmdgen.UdpTransportTarget(("demo.pysnmp.com", 161), timeout=5, retries=1)
oid = '1.3.6.1.2.1.1.1.0'
oid_tuple = tuple(int(part) for part in oid.split('.'))
comm_data = cmdgen.CommunityData('public', mpModel=0)
error_indication, error_status, error_index, var_binds = cmd_gen.getCmd(
    comm_data,
    transport,
    oid_tuple
)
for name, val in var_binds:
    print(f'{name} = {val}')

Other example:

from pysnmp.hlapi.v3arch.asyncio import *
import pysnmp_sync_adapter.cmdgen_wrappers as cmdgen
_oids = ('1.3.6.1.2.1.1.1.0', '1.3.6.1.2.1.1.4.0',
         '1.3.6.1.2.1.1.5.0', '1.3.6.1.2.1.1.6.0')
user = 'myUser'
authKe = 'authPassword'
privKe = 'privPassword'
authProto = usmHMACSHAAuthProtocol
privProto = usmAesCfb128Protocol
cmdGen = cmdgen.CommandGenerator()
cmdGen.getCmd(
    cmdgen.UsmUserData(
        user, authKey=authKe, privKey=privKe,
        authProtocol=authProto, privProtocol=privProto
    ),
    cmdgen.UdpTransportTarget(("demo.pysnmp.com", 161)),
    *[_ for _ in (ObjectType(ObjectIdentity(oid)) for oid in _oids)]
)

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 LICENCE for details.

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

1.0.8

Jun 15, 2025

1.0.7

Jun 7, 2025

1.0.6

Jun 2, 2025

1.0.5.post3

May 31, 2025

1.0.5.post2

May 31, 2025

1.0.5.post1

May 31, 2025

1.0.5

May 31, 2025

This version

1.0.4

May 26, 2025

1.0.3

May 26, 2025

1.0.2.post1

May 25, 2025

1.0.2

May 25, 2025

1.0.1

May 24, 2025

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pysnmp_sync_adapter-1.0.4.tar.gz (19.4 kB view details)

Uploaded May 26, 2025 Source

File details

Details for the file pysnmp_sync_adapter-1.0.4.tar.gz.

File metadata

Download URL: pysnmp_sync_adapter-1.0.4.tar.gz
Upload date: May 26, 2025
Size: 19.4 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.1.0 CPython/3.13.3

File hashes

Hashes for pysnmp_sync_adapter-1.0.4.tar.gz
Algorithm	Hash digest
SHA256	`dce3d412543eb7df212b940a0bf7fc9f0ef1cb5040e5a4a1a6e2b6a6fa64c0b2`
MD5	`4fcc7b656440eaab9a4786be587e338c`
BLAKE2b-256	`92e5b5f6b1a8bf9cc295ebf56b30e586ca484a34bc8bd86e296f398dea1dc3da`

See more details on using hashes here.

pysnmp-sync-adapter 1.0.4

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Project description

pysnmp-sync-adapter

Features

Provided Methods

Internal Utilities

Installation

Quick Start

Usage

High-level v1arch sync

High-level v3arch sync

Supporting other libraries

Support of the Python SNMP library v5.0.24 HLAPI

Example Usage

Support of legacy etingof/pysnmp v5 HLAPI

Example Usage

Notes

UdpTransportTarget

errorIndication

Support of the cmdgen SNMP library apearing in pysnmp oneliner

Example Usage

Notes and limitations

Reference documentation

License

Project details

Verified details

Maintainers

Unverified details

Project links

Meta

Release history Release notifications | RSS feed

Download files

Source Distribution

File details

File metadata

File hashes

Support of legacy `etingof/pysnmp` v5 HLAPI

`UdpTransportTarget`