pysnmp-sync-adapter 1.0.1

Synchronous wrapper adapters for pysnmp v1arch asyncio HLAPI

pysnmp-sync-adapter

Lightweight Synchronous Adapter for PySNMP v1arch AsyncIO HLAPI

---

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

Features

• Drop-in synchronous alternatives: get_cmd_sync, walk_cmd_sync, set_cmd_sync, and others.
• Reuses or creates a single shared event loop for efficiency.
• Pre-creates and reuses UdpTransportTarget to minimize connection overhead.

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 or creates the event loop, ensuring one loop per process
• create_transport() — pre-awaits UdpTransportTarget.create() once per host/port
• _sync_coro() — runs a coroutine to completion synchronously
• _sync_agen() — consumes async generators (e.g., for walk operations) into a list
• make_sync() — decorator to convert HLAPI coroutines into sync functions

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_sync_adapter import (
    get_cmd_sync, next_cmd_sync, set_cmd_sync, bulk_cmd_sync,
    walk_cmd_sync, bulk_walk_cmd_sync, create_transport
)
from pysnmp.hlapi.v1arch.asyncio import SnmpDispatcher, CommunityData, ObjectType, ObjectIdentity

dispatcher = SnmpDispatcher()
transport = create_transport('demo.pysnmp.com', 161)

err, status, index, var_binds = get_cmd_sync(
    dispatcher,
    CommunityData('public', mpModel=0),
    transport,
    ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))
)

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

Usage

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('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('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('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('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('demo.pysnmp.com', 161, timeout=2),
    ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr'))
)
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('demo.pysnmp.com', 161, timeout=2),
    0, 25,
    ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr')))
for error_indication, error_status, error_index, var_binds in objects:
    for name, val in var_binds:
        print(name.prettyPrint(), '=', val.prettyPrint())

Limitations

• These adapters block the calling thread until the SNMP operation completes.
• If the host app already drives an asyncio loop, calling these wrappers on that same loop can error or deadlock unless isolated (e.g. in a separate thread).
• The only way those calls give up on a slow or unresponsive SNMP peer is via the low-level socket own timeout; there’s no exposed mechanism to cancel the underlying asyncio task.

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.