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