Log input parameters of specific failing methods

src/zinolib/controllers/zino1.py

@@ -66,14 +66,15 @@
 """
 
 from datetime import datetime, timezone
-from typing import Iterable, List, TypedDict, Optional, Set
+from typing import Dict, Iterable, List, TypedDict, Optional, Set
 import logging
 
 from .base import EventManager
 from ..compat import StrEnum
 from ..event_types import EventType, Event, HistoryEntry, LogEntry, AdmState
 from ..event_types import PortStateEvent
 from ..ritz import ZinoError, ritz, notifier
+from ..utils import log_exception_with_params
 
 
 __all__ = [
@@ -118,11 +119,11 @@
 
     def __init__(self, manager, autoremove=False):
         if not manager.is_authenticated:
             msg = "Cannot initiate update handler, not authenticated"
             raise self.UpdateError(msg)
         self.manager = manager
         if not self.manager.session.push:
             self.manager.connect_push_channel()
         self.events = manager.events
         self.autoremove = autoremove
 
@@ -200,13 +201,13 @@
     @classmethod
     def connect_session(cls, session):
         session.request.connect()
         return cls.connect_push_channel(session)
 
     @staticmethod
     def connect_push_channel(session):
         if session.request.connected and session.request.authenticated:
             session.push = notifier(session.request)
             session.push.connect()  # ntie
         return session
 
     @staticmethod
@@ -220,8 +221,8 @@
 
     @staticmethod
     def close_session(session):
         if hasattr(session.push, '_sock'):
             session.push._sock.close()
         session.request.close()
         return None
 
@@ -248,7 +249,8 @@
         return request.get_raw_attributes(event_id)
 
     @classmethod
-    def attrlist_to_attrdict(cls, attrlist: Iterable[str]):
+    @log_exception_with_params(LOG)
+    def attrlist_to_attrdict(cls, attrlist: Iterable[str]) -> Dict[str, str]:
         """Translate a wire protocol dump of a single event
 
         The dump is a list of lines of the format:
@@ -290,6 +292,7 @@
         return request.get_raw_history(event_id).data
 
     @classmethod
+    @log_exception_with_params(LOG)
     def parse_response(cls, history_data: Iterable[str]) -> list[HistoryDict]:
         """
         Input:
@@ -357,6 +360,7 @@
         return request.get_raw_log(event_id).data
 
     @staticmethod
+    @log_exception_with_params(LOG)
     def parse_response(log_data: Iterable[str]) -> list[LogDict]:
         """
         Input:
@@ -404,7 +408,7 @@
         "Replace the original exception with our own"
         try:
             return function(*args)
         except ZinoError as e:
             raise self.ManagerException(e)
 
     def _verify_session(self, quiet=False):
@@ -424,12 +428,12 @@
         self.session = self._session_adapter.connect_session(self.session)
 
     def connect_push_channel(self):
         self.session = self._session_adapter.connect_push_channel(self.session)
 
     def authenticate(self, username=None, password=None):
         try:
             self.session = self._session_adapter.authenticate(self.session, username, password)
         except (ZinoError, ValueError) as e:
             raise self.ManagerException(e)
 
     def disconnect(self):
@@ -444,7 +448,7 @@
             c.clear_flapping()
         """
         if event.type == Event.Type.PortState:
             return self.session.request.clear_flapping(event.router, event.if_index)
         return None
 
     def get_events(self):

src/zinolib/utils.py

@@ -1,4 +1,6 @@
+import functools
 import hashlib
+from typing import Any
 
 
 __all__ = [
@@ -56,3 +58,26 @@
     raw_token = "%s %s" % (challenge, password)
     token = hashlib.sha1(raw_token.encode("UTF-8")).hexdigest()
     return token
+
+
+def log_exception_with_params(logger, reraise=True, return_value=None):
+    """Log the params and exception if the wrapped function fails
+
+    If ``reraise`` is False, return ``return_value`` instead of reraising the
+    exception.
+    """
+    def inner(function):
+        @functools.wraps(function)
+        def wrapper(*args, **kwargs):
+            try:
+                result = function(*args, **kwargs)
+                return result
+            except Exception as e:
+                params = f'args={args} kwargs={kwargs}'
+                funcname = function.__name__
+                logger.exception(f'"{funcname}" failed with: {params}\n{e}')
+                if reraise:
+                    raise
+                return return_value
+        return wrapper
+    return inner