puresnmp.transport module

Low-Level network transport for asyncio.

This module mainly exist to enable a “seam” for mocking/patching out during testing.

The module is excluded from coverage. It contains all the “dirty” stuff that’s hard to test.

class puresnmp.transport.Endpoint(ip: IPv4Address | IPv6Address, port: int)

Bases: NamedTuple

A tuple representing an UDP endpoint where a connection should be made to.

ip: IPv4Address | IPv6Address

Alias for field number 0

port: int

Alias for field number 1

class puresnmp.transport.SNMPClientProtocol(packet: bytes)

Bases: DatagramProtocol

An SNMP Client Protocol suitable for use with asyncio.AbstractEventLoop.create_datagram_endpoint() that provides a method to convert the callback based API into a coroutine based API.

connection_lost(exc: Exception | None) → None

Handles the socket being closed optionally passing on an exception.

connection_made(transport: DatagramTransport) → None

Sends the SNMP request packet when a connection is made.

datagram_received(data: bytes | str, addr: Tuple[str, int]) → None

Receive the data and close the connection.

error_received(exc: Exception) → None

Pass the exception along if there is an error.

async get_data(timeout: int) → bytes

Retrieve the response data back into the calling coroutine.

class puresnmp.transport.SNMPTrapReceiverProtocol(callback: Callable[[SocketResponse], Any])

Bases: DatagramProtocol

A protocol to handle incoming SNMP traps.

The protocol requires a callable which is called with a SocketResponse instance whenever a trap is received.

connection_made(transport: BaseTransport) → None

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

datagram_received(data: bytes, addr: Tuple[str, int]) → None

Called when some datagram is received.

class puresnmp.transport.TSender(*args, **kwargs)

Bases: Protocol

A typing-protocol for callables which send data out to the network

puresnmp.transport.default_trap_handler(response: SocketResponse) → None

A no-op implementation for trap handlers which only logs traps.

async puresnmp.transport.listen(bind_address: str = '0.0.0.0', port: int = 162, callback: ~typing.Callable[[~puresnmp.typevars.SocketResponse], ~typing.Any] = <function default_trap_handler>, loop: ~asyncio.events.AbstractEventLoop | None = None) → None

Sets up a listening UDP socket and returns a generator over recevied packets:

>>> transport = Transport()  
>>> for seq, packet in enumerate(transport.listen()):  
...     print(seq, repr(packet))
0, b'...'
1, b'...'
2, b'...'

Note

This defaults to the standard SNMP Trap port 162. This is a privileged port so processes using this port must run as root!

async puresnmp.transport.send_udp(endpoint: Endpoint, packet: bytes, timeout: int = 1, loop: AbstractEventLoop | None = None, retries: int = 10) → bytes

A coroutine that opens a UDP socket to ip:port, sends a packet with bytes and returns the raw bytes as returned from the remote host.

If the connection fails due to a timeout, a Timeout exception is raised.