Shorten ObjectIdentifier.from_string calls
@@ -41,6 +41,7 @@ except NameError:
_set = set
LOG = logging.getLogger(__name__)
+OID = ObjectIdentifier.from_string
def get(ip, community, oid, port=161, timeout=2):
@@ -70,7 +71,7 @@ def multiget(ip, community, oids, port=161, timeout=2):
['non-functional example', 'second value']
"""
- oids = [ObjectIdentifier.from_string(oid) for oid in oids]
+ oids = [OID(oid) for oid in oids]
packet = Sequence(
Integer(Version.V2C),
@@ -175,7 +176,7 @@ def multiwalk(ip, community, oids, port=161, timeout=2, fetcher=multigetnext):
LOG.debug('Walking on %d OIDs using %s', len(oids), fetcher.__name__)
varbinds = fetcher(ip, community, oids, port, timeout)
- requested_oids = [ObjectIdentifier.from_string(oid) for oid in oids]
+ requested_oids = [OID(oid) for oid in oids]
grouped_oids = group_varbinds(varbinds, requested_oids)
unfinished_oids = get_unfinished_walk_oids(grouped_oids)
LOG.debug('%d of %d OIDs need to be continued',
@@ -254,7 +255,7 @@ def multiset(ip, community, mappings, port=161, timeout=2):
raise TypeError('SNMP requires typing information. The value for a '
'"set" request must be an instance of "Type"!')
- binds = [VarBind(ObjectIdentifier.from_string(k), v)
+ binds = [VarBind(OID(k), v)
for k, v in mappings]
request = SetRequest(get_request_id(), binds)
@@ -343,9 +344,9 @@ def bulkget(ip, community, scalar_oids, repeating_oids, max_list_size=1,
repeating_oids = repeating_oids or [] # protect against empty values
oids = [
- ObjectIdentifier.from_string(oid) for oid in scalar_oids
+ OID(oid) for oid in scalar_oids
] + [
- ObjectIdentifier.from_string(oid) for oid in repeating_oids
+ OID(oid) for oid in repeating_oids
]
non_repeaters = len(scalar_oids)
@@ -393,14 +394,15 @@ def _bulkwalk_fetcher(bulk_size=10):
"""
Create a bulk fetcher with a fixed limit on "repeatable" OIDs.
"""
+
def fetcher(ip, community, oids, port=161, timeout=2):
'''
Executes a SNMP BulkGet request.
'''
result = bulkget(ip, community, [], oids, max_list_size=bulk_size,
port=port, timeout=timeout)
- return [VarBind(ObjectIdentifier.from_string(k), v)
- for k, v in result.listing.items()]
+ return [VarBind(OID(k), v) for k, v in result.listing.items()]
+
if sys.version_info < (3, 0):
fetcher.__name__ = str('_bulkwalk_fetcher(%d)' % bulk_size)
else:
