mirror of
https://github.com/checktheroads/hyperglass
synced 2024-05-11 05:55:08 +00:00
WIP: improve exceptions & error handling
This commit is contained in:
@@ -2,8 +2,7 @@
|
|||||||
Custom exceptions for hyperglass
|
Custom exceptions for hyperglass
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import string
|
from typing import Dict
|
||||||
from typing import Union, List
|
|
||||||
|
|
||||||
from hyperglass.constants import code
|
from hyperglass.constants import code
|
||||||
|
|
||||||
@@ -13,61 +12,7 @@ class HyperglassError(Exception):
|
|||||||
hyperglass base exception
|
hyperglass base exception
|
||||||
"""
|
"""
|
||||||
|
|
||||||
message: str = ""
|
pass
|
||||||
formatter: string.Formatter = string.Formatter()
|
|
||||||
|
|
||||||
def __init__(self, **kwargs: Union[str, int]) -> None:
|
|
||||||
"""
|
|
||||||
Exception arguments are accepted as kwargs, but a check is
|
|
||||||
performed to ensure that all format string parameters are passed
|
|
||||||
in, and no extras are given.
|
|
||||||
"""
|
|
||||||
self._kwargs = kwargs
|
|
||||||
self._error_check()
|
|
||||||
super().__init__(str(self))
|
|
||||||
|
|
||||||
def _error_check(self) -> None:
|
|
||||||
required = set(
|
|
||||||
arg for _, arg, _, _ in self.formatter.parse(self.message) if arg
|
|
||||||
)
|
|
||||||
given = set(self._kwargs.keys())
|
|
||||||
|
|
||||||
missing = required.difference(given)
|
|
||||||
if missing:
|
|
||||||
raise TypeError(
|
|
||||||
"{name} missing requred arguments: {missing}".format(
|
|
||||||
name=self.__class__.__name__, missing=missing
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
extra = given.difference(required)
|
|
||||||
if extra:
|
|
||||||
raise TypeError(
|
|
||||||
"{name} given extra arguments: {extra}".format(
|
|
||||||
name=self.__class__.__name__, extra=extra
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
def __str__(self) -> str:
|
|
||||||
return self.formatter.format(self.message, **self._kwargs)
|
|
||||||
|
|
||||||
def __getattr__(self, key: str) -> str:
|
|
||||||
"""
|
|
||||||
Any exception kwargs arguments are accessible by name on the
|
|
||||||
object.
|
|
||||||
"""
|
|
||||||
remind = ""
|
|
||||||
if "_kwargs" not in self.__dict__:
|
|
||||||
remind = "(Did you forget to call super().__init__(**kwargs)?)"
|
|
||||||
|
|
||||||
elif key in self._kwargs:
|
|
||||||
return self._kwargs[key]
|
|
||||||
|
|
||||||
raise AttributeError(
|
|
||||||
"{name!r} object has no attribute {key!r} {remind}".format(
|
|
||||||
name=self.__class__.__name__, key=key, remind=remind
|
|
||||||
).strip()
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class ConfigError(HyperglassError):
|
class ConfigError(HyperglassError):
|
||||||
@@ -75,13 +20,27 @@ class ConfigError(HyperglassError):
|
|||||||
Raised for generic user-config issues.
|
Raised for generic user-config issues.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
message: str = "{error_msg}"
|
def __init__(self, unformatted_msg, kwargs={}):
|
||||||
|
self.message: unformatted_msg.format(**kwargs)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class ConfigInvalid(HyperglassError):
|
class ConfigInvalid(HyperglassError):
|
||||||
"""Raised when a config item fails type or option validation"""
|
"""Raised when a config item fails type or option validation"""
|
||||||
|
|
||||||
message: str = 'The value field "{field}" is invalid: {error_msg}'
|
def __init__(self, **kwargs):
|
||||||
|
self.message: str = 'The value field "{field}" is invalid: {error_msg}'.format(
|
||||||
|
**kwargs
|
||||||
|
)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class ConfigMissing(HyperglassError):
|
class ConfigMissing(HyperglassError):
|
||||||
@@ -89,50 +48,65 @@ class ConfigMissing(HyperglassError):
|
|||||||
Raised when a required config file or item is missing or undefined
|
Raised when a required config file or item is missing or undefined
|
||||||
"""
|
"""
|
||||||
|
|
||||||
message: str = (
|
def __init__(self, kwargs={}):
|
||||||
|
self.message: str = (
|
||||||
"{missing_item} is missing or undefined and is required to start "
|
"{missing_item} is missing or undefined and is required to start "
|
||||||
"hyperglass. Please consult the installation documentation."
|
"hyperglass. Please consult the installation documentation."
|
||||||
)
|
).format(**kwargs)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class ScrapeError(HyperglassError):
|
class ScrapeError(HyperglassError):
|
||||||
"""Raised upon a scrape/netmiko error"""
|
"""Raised upon a scrape/netmiko error"""
|
||||||
|
|
||||||
message: str = ""
|
def __init__(self, kwargs={}):
|
||||||
status: int = code.target_error
|
self.message: str = "".format(**kwargs)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.target_error
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class AuthError(HyperglassError):
|
class AuthError(HyperglassError):
|
||||||
"""Raised when authentication to a device fails"""
|
"""Raised when authentication to a device fails"""
|
||||||
|
|
||||||
message: str = ""
|
def __init__(self, kwargs={}):
|
||||||
status: int = code.target_error
|
self.message: str = "".format(**kwargs)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.target_error
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class RestError(HyperglassError):
|
class RestError(HyperglassError):
|
||||||
"""Raised upon a rest API client error"""
|
"""Raised upon a rest API client error"""
|
||||||
|
|
||||||
message: str = ""
|
def __init__(self, kwargs={}):
|
||||||
status: int = code.target_error
|
self.message: str = "".format(**kwargs)
|
||||||
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.target_error
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class InputInvalid(HyperglassError):
|
class InputInvalid(HyperglassError):
|
||||||
"""Raised when input validation fails"""
|
"""Raised when input validation fails"""
|
||||||
|
|
||||||
def __init__(self, **kwargs):
|
def __init__(self, unformatted_msg, **kwargs):
|
||||||
self._kwargs = kwargs
|
self.message: str = unformatted_msg.format(**kwargs)
|
||||||
self._query_type = self._kwargs.get("query_type")
|
self.keywords: Dict = kwargs
|
||||||
if self._query_type in ("bgp_route", "ping", "traceroute"):
|
|
||||||
self.query_type: str = "IP Address"
|
|
||||||
elif self._query_type == "bgp_aspath":
|
|
||||||
self.query_type: str = "AS Path"
|
|
||||||
elif self._query_type == "bgp_community":
|
|
||||||
self.query_type: str = "Community"
|
|
||||||
self.target: str = str(kwargs.get("target"), None)
|
|
||||||
self.message = f"{self.target} is an invalid {self.query_type}."
|
|
||||||
self.status: int = code.invalid
|
self.status: int = code.invalid
|
||||||
self.keywords: List[str] = []
|
super().__init__(self.message, self.status)
|
||||||
super().__init__(self.message, self.status, self.keywords)
|
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return self.message
|
return self.message
|
||||||
@@ -144,9 +118,14 @@ class InputNotAllowed(HyperglassError):
|
|||||||
requires_ipv6_cidr check
|
requires_ipv6_cidr check
|
||||||
"""
|
"""
|
||||||
|
|
||||||
message: str = ""
|
def __init__(self, unformatted_msg, **kwargs):
|
||||||
status: int = code.not_allowed
|
self.message: str = unformatted_msg.format(**kwargs)
|
||||||
keywords: List[str] = []
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.invalid
|
||||||
|
super().__init__(self.status, self.message)
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.message
|
||||||
|
|
||||||
|
|
||||||
class ParseError(HyperglassError):
|
class ParseError(HyperglassError):
|
||||||
@@ -154,9 +133,11 @@ class ParseError(HyperglassError):
|
|||||||
Raised when an ouput parser encounters an error.
|
Raised when an ouput parser encounters an error.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, message):
|
def __init__(self, kwargs={}):
|
||||||
super().__init__(message)
|
self.message: str = "".format(**kwargs)
|
||||||
self.message = message
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.target_error
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return self.message
|
return self.message
|
||||||
@@ -167,9 +148,11 @@ class UnsupportedDevice(HyperglassError):
|
|||||||
Raised when an input NOS is not in the supported NOS list.
|
Raised when an input NOS is not in the supported NOS list.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, message):
|
def __init__(self, kwargs={}):
|
||||||
super().__init__(message)
|
self.message: str = "".format(**kwargs)
|
||||||
self.message = message
|
self.keywords: Dict = kwargs
|
||||||
|
self.status: int = code.target_error
|
||||||
|
super().__init__(self.message, self.keywords)
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return self.message
|
return self.message
|
||||||
|
Reference in New Issue
Block a user