Python v3.x: exception handler for coroutine and synchronous functions. In general, for everything

In my free time I work on my small project . Written in Python v3.x + SQLAlchemy. Maybe I will write about it someday, but today I want to talk about my decorator for handling exceptions. It can be used both for functions and for methods. Synchronous and asynchronous. You can also connect custom exception handlers.



The decorator currently looks like this:

import asyncio from asyncio import QueueEmpty, QueueFull from concurrent.futures import TimeoutError class ProcessException(object): __slots__ = ('func', 'custom_handlers', 'exclude') def __init__(self, custom_handlers=None): self.func = None self.custom_handlers: dict = custom_handlers self.exclude = [QueueEmpty, QueueFull, TimeoutError] def __call__(self, func, *a): self.func = func def wrapper(*args, **kwargs): if self.custom_handlers: if isinstance(self.custom_handlers, property): self.custom_handlers = self.custom_handlers.__get__(self, self.__class__) if asyncio.iscoroutinefunction(self.func): return self._coroutine_exception_handler(*args, **kwargs) else: return self._sync_exception_handler(*args, **kwargs) return wrapper async def _coroutine_exception_handler(self, *args, **kwargs): try: return await self.func(*args, **kwargs) except Exception as e: if self.custom_handlers and e.__class__ in self.custom_handlers: return self.custom_handlers[e.__class__]() if e.__class__ not in self.exclude: raise e def _sync_exception_handler(self, *args, **kwargs): try: return self.func(*args, **kwargs) except Exception as e: if self.custom_handlers and e.__class__ in self.custom_handlers: return self.custom_handlers[e.__class__]() if e.__class__ not in self.exclude: raise e
      
      





We will analyze in order. __slots__ I use to save a little memory. It is useful if the object is used sooo often.



At the initialization stage in __init__, we save custom_handlers (in case it was necessary to transfer them). Just in case, he indicated that we expect to see a dictionary there, although, perhaps in the future, it makes sense to add a couple of hard checks. The self.exclude property contains a list of exceptions that you do not need to handle. In case of such an exception, the function with the decorator will return None. At the moment, the list is sharpened for my project, and perhaps it makes sense to put it in a separate config.



The most important thing happens in __call__. Therefore, when using the decorator, you need to call it. Even without parameters:



 @ProcessException() def some_function(*args): return None
      
      





Those. this will be wrong and an error will be raised:



 @ProcessException def some_function(*args): return None
      
      





In this case, we get the current function, which, depending on the degree of its asynchrony, we will process either as a regular synchronous or as a coroutine.



What you can pay attention to here. The first is a check on the property:



 if self.custom_handlers: if isinstance(self.custom_handlers, property): self.custom_handlers = self.custom_handlers.__get__(self, self.__class__)
      
      





Why am I doing this?



 Of course 
          not because 
                       I'm IT Mayakovsky 
                                    and they pay me line by line.


Two if are here to improve readability (yes, the code can be supported by a person with sadistic tendencies), and we do self.custom_handlers .__ get __ (self, self .__ class__) in case we decide handlers stored in the @property class.



For example, like this:



 class Math(object): @property def exception_handlers(self): return { ZeroDivisionError: lambda: '   ,   ' } @ProcessException(exception_handlers) def divide(self, a, b): return a // b
      
      





If we do not do self.custom_handlers .__ get __ (...), then instead of the contents of @property we will get something like <property object at 0x7f78d844f9b0>.



Actually, the example above shows how to connect custom handlers. In general, this is done as follows:



 @ProcessException({ZeroDivisionError: lambda: '   ,   '}) def divide(a, b): return a // b
      
      





In the case of a class (if we are going to pass properties / methods), we need to take into account that at the stage of initializing the class decorator , there are no such methods and the methods / properties are simple functions. Therefore, we can only convey what is announced above. Therefore, the @property option is the ability to use through self all the functions that are lower in code. Well, either you can use lambdas if self is not needed.



For asynchronous code, all the above examples are true.



Finally, I want to draw attention to the fact that if an exception in its path did not meet custom handlers, then it simply raises further.



Waiting for your comments. Thank you for paying attention to my article.



All Articles