Loading src/device/service/driver_api/DriverFactory.py +15 −9 Original line number Diff line number Diff line Loading @@ -14,8 +14,7 @@ import logging from enum import Enum from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Type from ._Driver import _Driver from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple, Type from .Exceptions import ( AmbiguousFilterException, EmptyFilterFieldException, UnsatisfiedFilterException, UnsupportedDriverClassException, Loading @@ -23,12 +22,20 @@ from .Exceptions import ( ) from .FilterFields import FILTER_FIELD_ALLOWED_VALUES, FilterFieldEnum if TYPE_CHECKING: from ._Driver import _Driver LOGGER = logging.getLogger(__name__) SUPPORTED_FILTER_FIELDS = set(FILTER_FIELD_ALLOWED_VALUES.keys()) def check_is_class_valid(driver_class : Type['_Driver']) -> None: from ._Driver import _Driver if not issubclass(driver_class, _Driver): raise UnsupportedDriverClassException(str(driver_class)) def sanitize_filter_fields( filter_fields : Dict[FilterFieldEnum, Any], driver_name : Optional[str] = None ) -> Dict[FilterFieldEnum, Any]: Loading Loading @@ -67,14 +74,13 @@ def sanitize_filter_fields( class DriverFactory: def __init__( self, drivers : List[Tuple[Type[_Driver], List[Dict[FilterFieldEnum, Any]]]] self, drivers : List[Tuple[Type['_Driver'], List[Dict[FilterFieldEnum, Any]]]] ) -> None: self.__drivers : List[Tuple[Type[_Driver], Dict[FilterFieldEnum, Any]]] = list() self.__drivers : List[Tuple[Type['_Driver'], Dict[FilterFieldEnum, Any]]] = list() for driver_class,filter_field_sets in drivers: #if not issubclass(driver_class, _Driver): # raise UnsupportedDriverClassException(str(driver_class)) driver_name = driver_class #.__name__ check_is_class_valid(driver_class) driver_name = driver_class.__name__ for filter_fields in filter_field_sets: filter_fields = {k.value:v for k,v in filter_fields.items()} Loading Loading @@ -102,7 +108,7 @@ class DriverFactory: return True def get_driver_class(self, **selection_filter_fields) -> _Driver: def get_driver_class(self, **selection_filter_fields) -> '_Driver': sanitized_filter_fields = sanitize_filter_fields(selection_filter_fields) compatible_drivers : List[Tuple[Type[_Driver], Dict[FilterFieldEnum, Any]]] = [ Loading src/device/service/driver_api/Exceptions.py +12 −10 Original line number Diff line number Diff line Loading @@ -13,22 +13,22 @@ # limitations under the License. class UnsatisfiedFilterException(Exception): def __init__(self, filter_fields): def __init__(self, filter_fields) -> None: msg = 'No Driver satisfies FilterFields({:s})' super().__init__(msg.format(str(filter_fields))) class AmbiguousFilterException(Exception): def __init__(self, filter_fields, compatible_drivers): def __init__(self, filter_fields, compatible_drivers) -> None: msg = 'Multiple Drivers satisfy FilterFields({:s}): {:s}' super().__init__(msg.format(str(filter_fields), str(compatible_drivers))) class UnsupportedDriverClassException(Exception): def __init__(self, driver_class_name): def __init__(self, driver_class_name) -> None: msg = 'Class({:s}) is not a subclass of _Driver' super().__init__(msg.format(str(driver_class_name))) class EmptyFilterFieldException(Exception): def __init__(self, filter_fields, driver_class_name=None): def __init__(self, filter_fields, driver_class_name=None) -> None: if driver_class_name: msg = 'Empty FilterField({:s}) specified by Driver({:s}) is not supported' msg = msg.format(str(filter_fields), str(driver_class_name)) Loading @@ -38,7 +38,7 @@ class EmptyFilterFieldException(Exception): super().__init__(msg) class UnsupportedFilterFieldException(Exception): def __init__(self, unsupported_filter_fields, driver_class_name=None): def __init__(self, unsupported_filter_fields, driver_class_name=None) -> None: if driver_class_name: msg = 'FilterFields({:s}) specified by Driver({:s}) are not supported' msg = msg.format(str(unsupported_filter_fields), str(driver_class_name)) Loading @@ -48,7 +48,9 @@ class UnsupportedFilterFieldException(Exception): super().__init__(msg) class UnsupportedFilterFieldValueException(Exception): def __init__(self, filter_field_name, filter_field_value, allowed_filter_field_values, driver_class_name=None): def __init__( self, filter_field_name, filter_field_value, allowed_filter_field_values, driver_class_name=None ) -> None: if driver_class_name: msg = 'FilterField({:s}={:s}) specified by Driver({:s}) is not supported. Allowed values are {:s}' msg = msg.format( Loading @@ -60,24 +62,24 @@ class UnsupportedFilterFieldValueException(Exception): super().__init__(msg) class DriverInstanceCacheTerminatedException(Exception): def __init__(self): def __init__(self) -> None: msg = 'DriverInstanceCache is terminated. No new instances can be processed.' super().__init__(msg) class UnsupportedResourceKeyException(Exception): def __init__(self, resource_key): def __init__(self, resource_key) -> None: msg = 'ResourceKey({:s}) not supported' msg = msg.format(str(resource_key)) super().__init__(msg) class ConfigFieldNotFoundException(Exception): def __init__(self, config_field_name): def __init__(self, config_field_name) -> None: msg = 'ConfigField({:s}) not specified in resource' msg = msg.format(str(config_field_name)) super().__init__(msg) class ConfigFieldsNotSupportedException(Exception): def __init__(self, config_fields): def __init__(self, config_fields) -> None: msg = 'ConfigFields({:s}) not supported in resource' msg = msg.format(str(config_fields)) super().__init__(msg) Loading
src/device/service/driver_api/DriverFactory.py +15 −9 Original line number Diff line number Diff line Loading @@ -14,8 +14,7 @@ import logging from enum import Enum from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Type from ._Driver import _Driver from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple, Type from .Exceptions import ( AmbiguousFilterException, EmptyFilterFieldException, UnsatisfiedFilterException, UnsupportedDriverClassException, Loading @@ -23,12 +22,20 @@ from .Exceptions import ( ) from .FilterFields import FILTER_FIELD_ALLOWED_VALUES, FilterFieldEnum if TYPE_CHECKING: from ._Driver import _Driver LOGGER = logging.getLogger(__name__) SUPPORTED_FILTER_FIELDS = set(FILTER_FIELD_ALLOWED_VALUES.keys()) def check_is_class_valid(driver_class : Type['_Driver']) -> None: from ._Driver import _Driver if not issubclass(driver_class, _Driver): raise UnsupportedDriverClassException(str(driver_class)) def sanitize_filter_fields( filter_fields : Dict[FilterFieldEnum, Any], driver_name : Optional[str] = None ) -> Dict[FilterFieldEnum, Any]: Loading Loading @@ -67,14 +74,13 @@ def sanitize_filter_fields( class DriverFactory: def __init__( self, drivers : List[Tuple[Type[_Driver], List[Dict[FilterFieldEnum, Any]]]] self, drivers : List[Tuple[Type['_Driver'], List[Dict[FilterFieldEnum, Any]]]] ) -> None: self.__drivers : List[Tuple[Type[_Driver], Dict[FilterFieldEnum, Any]]] = list() self.__drivers : List[Tuple[Type['_Driver'], Dict[FilterFieldEnum, Any]]] = list() for driver_class,filter_field_sets in drivers: #if not issubclass(driver_class, _Driver): # raise UnsupportedDriverClassException(str(driver_class)) driver_name = driver_class #.__name__ check_is_class_valid(driver_class) driver_name = driver_class.__name__ for filter_fields in filter_field_sets: filter_fields = {k.value:v for k,v in filter_fields.items()} Loading Loading @@ -102,7 +108,7 @@ class DriverFactory: return True def get_driver_class(self, **selection_filter_fields) -> _Driver: def get_driver_class(self, **selection_filter_fields) -> '_Driver': sanitized_filter_fields = sanitize_filter_fields(selection_filter_fields) compatible_drivers : List[Tuple[Type[_Driver], Dict[FilterFieldEnum, Any]]] = [ Loading
src/device/service/driver_api/Exceptions.py +12 −10 Original line number Diff line number Diff line Loading @@ -13,22 +13,22 @@ # limitations under the License. class UnsatisfiedFilterException(Exception): def __init__(self, filter_fields): def __init__(self, filter_fields) -> None: msg = 'No Driver satisfies FilterFields({:s})' super().__init__(msg.format(str(filter_fields))) class AmbiguousFilterException(Exception): def __init__(self, filter_fields, compatible_drivers): def __init__(self, filter_fields, compatible_drivers) -> None: msg = 'Multiple Drivers satisfy FilterFields({:s}): {:s}' super().__init__(msg.format(str(filter_fields), str(compatible_drivers))) class UnsupportedDriverClassException(Exception): def __init__(self, driver_class_name): def __init__(self, driver_class_name) -> None: msg = 'Class({:s}) is not a subclass of _Driver' super().__init__(msg.format(str(driver_class_name))) class EmptyFilterFieldException(Exception): def __init__(self, filter_fields, driver_class_name=None): def __init__(self, filter_fields, driver_class_name=None) -> None: if driver_class_name: msg = 'Empty FilterField({:s}) specified by Driver({:s}) is not supported' msg = msg.format(str(filter_fields), str(driver_class_name)) Loading @@ -38,7 +38,7 @@ class EmptyFilterFieldException(Exception): super().__init__(msg) class UnsupportedFilterFieldException(Exception): def __init__(self, unsupported_filter_fields, driver_class_name=None): def __init__(self, unsupported_filter_fields, driver_class_name=None) -> None: if driver_class_name: msg = 'FilterFields({:s}) specified by Driver({:s}) are not supported' msg = msg.format(str(unsupported_filter_fields), str(driver_class_name)) Loading @@ -48,7 +48,9 @@ class UnsupportedFilterFieldException(Exception): super().__init__(msg) class UnsupportedFilterFieldValueException(Exception): def __init__(self, filter_field_name, filter_field_value, allowed_filter_field_values, driver_class_name=None): def __init__( self, filter_field_name, filter_field_value, allowed_filter_field_values, driver_class_name=None ) -> None: if driver_class_name: msg = 'FilterField({:s}={:s}) specified by Driver({:s}) is not supported. Allowed values are {:s}' msg = msg.format( Loading @@ -60,24 +62,24 @@ class UnsupportedFilterFieldValueException(Exception): super().__init__(msg) class DriverInstanceCacheTerminatedException(Exception): def __init__(self): def __init__(self) -> None: msg = 'DriverInstanceCache is terminated. No new instances can be processed.' super().__init__(msg) class UnsupportedResourceKeyException(Exception): def __init__(self, resource_key): def __init__(self, resource_key) -> None: msg = 'ResourceKey({:s}) not supported' msg = msg.format(str(resource_key)) super().__init__(msg) class ConfigFieldNotFoundException(Exception): def __init__(self, config_field_name): def __init__(self, config_field_name) -> None: msg = 'ConfigField({:s}) not specified in resource' msg = msg.format(str(config_field_name)) super().__init__(msg) class ConfigFieldsNotSupportedException(Exception): def __init__(self, config_fields): def __init__(self, config_fields) -> None: msg = 'ConfigFields({:s}) not supported in resource' msg = msg.format(str(config_fields)) super().__init__(msg)