File: //usr/local/bin/dhwp/env/lib/python3.10/site-packages/cement/core/handler.py
"""
Cement core handler module.
"""
from __future__ import annotations
import re
from abc import ABC
from typing import Any, List, Dict, Optional, Type, Union, TYPE_CHECKING
from ..core import exc
from ..core.meta import MetaMixin
from ..utils.misc import minimal_logger
LOG = minimal_logger(__name__)
if TYPE_CHECKING:
from ..core.foundation import App # pragma: nocover
class Handler(ABC, MetaMixin):
"""Base handler class that all Cement Handlers should subclass from."""
class Meta:
"""
Handler meta-data (can also be passed as keyword arguments to the
parent class).
"""
label: str = NotImplemented
"""The string identifier of this handler."""
interface: str = NotImplemented
"""The interface that this class implements."""
config_section: str = None # type: ignore
"""
A config section to merge config_defaults with.
Note: Though ``App.Meta.config_section`` defaults to ``None``, Cement
will set this to the value of ``<interface_label>.<handler_label>`` if
no section is set by the user/developer.
"""
config_defaults: Optional[Dict[str, Any]] = None
"""
A config dictionary that is merged into the applications config
in the ``[<config_section>]`` block. These are defaults and do not
override any existing defaults under that section.
"""
overridable: bool = False
"""
Whether or not handler can be overridden by
``App.Meta.handler_override_options``. Will be listed as an
available choice to override the specific handler (i.e.
``App.Meta.output_handler``, etc).
"""
def __init__(self, **kw: Any) -> None:
super(Handler, self).__init__(**kw)
try:
assert self._meta.label, \
f"{self.__class__.__name__}.Meta.label undefined."
assert self._meta.interface, \
f"{self.__class__.__name__}.Meta.interface undefined."
except AssertionError as e:
raise exc.FrameworkError(e.args[0])
self.app: App = None # type: ignore
def _setup(self, app: App) -> None:
"""
Called during application initialization and must ``setup`` the handler
object making it ready for the framework or the application to make
further calls to it.
Args:
app (instance): The application object.
"""
self.app = app
if self._meta.config_section is None:
self._meta.config_section = f"{self._meta.interface}.{self._meta.label}"
if self._meta.config_defaults is not None:
LOG.debug(f"merging config defaults from '{self}' " +
f"into section '{self._meta.config_section}'")
dict_obj = dict()
dict_obj[self._meta.config_section] = self._meta.config_defaults
self.app.config.merge(dict_obj, override=False)
self._validate()
def _validate(self) -> None:
"""
Perform any validation to ensure proper data, meta-data, etc.
"""
pass # pragma: nocover
class HandlerManager(object):
"""
Manages the handler system to define, get, resolve, etc handlers with
the Cement Framework.
"""
def __init__(self, app: App):
self.app = app
self.__handlers__: Dict[str, dict[str, Type[Handler]]] = {}
def get(self,
interface: str,
handler_label: str,
fallback: Optional[Type[Handler]] = None,
**kwargs: Any) -> Union[Handler, Type[Handler]]:
"""
Get a handler object.
Args:
interface (str): The interface of the handler (i.e. ``output``)
handler_label (str): The label of the handler (i.e. ``json``)
fallback (Handler): A fallback value to return if handler_label
doesn't exist.
Keyword Args:
setup (bool): Whether or not to call ``setup()`` on the handler
before returning. This will not be called on the ``fallback``
if no the handler given does not exist.
Returns:
Handler: An uninstantiated handler object
Raises:
cement.core.exc.InterfaceError: If the ``interface`` does not
exist, or if the handler itself does not exist.
Example:
.. code-block:: python
_handler = app.handler.get('output', 'json')
output = _handler()
output._setup(app)
output.render(dict(foo='bar'))
"""
setup = kwargs.get('setup', False)
if interface not in self.app.interface.list():
raise exc.InterfaceError(f"Interface '{interface}' does not exist!")
if handler_label in self.__handlers__[interface]:
if setup is True:
han = self.__handlers__[interface][handler_label]
return self.setup(han)
else:
return self.__handlers__[interface][handler_label]
elif fallback is not None:
return fallback
else:
raise exc.InterfaceError("handlers['%s']['%s'] does not exist!" %
(interface, handler_label))
def list(self, interface: str) -> List[Type[Handler]]:
"""
Return a list of handlers for a given ``interface``.
Args:
interface (str): The interface of the handler (i.e. ``output``)
Returns:
list: Handler labels (str) that match ``interface``.
Raises:
cement.core.exc.InterfaceError: If the ``interface`` does not
exist.
Example:
.. code-block:: python
app.handler.list('log')
"""
if not self.app.interface.defined(interface):
raise exc.InterfaceError(f"Interface '{interface}' does not exist!")
res = []
for label in self.__handlers__[interface]:
res.append(self.__handlers__[interface][label])
return res
def register(self,
handler_class: Type[Handler],
force: bool = False) -> None:
"""
Register a handler class to an interface. If the same object is
already registered then no exception is raised, however if a different
object attempts to be registered to the same name a ``InterfaceError``
is raised.
Args:
handler_class (Handler): The uninstantiated handler class to
register.
Keyword Arguments:
force (bool): Whether to allow replacement if an existing
handler of the same ``label`` is already registered.
Raises:
cement.core.exc.InterfaceError: If the ``handler_class`` does not
implement :class:`Handler`, or if ``handler_class`` does not
properly sub-class it's interface.
cement.core.exc.InterfaceError: If the
``handler_class.Meta.interface`` does not exist
Usage:
.. code-block:: python
class MyDatabaseHandler(object):
class Meta:
interface = IDatabase
label = 'mysql'
def connect(self):
# ...
app.handler.register(MyDatabaseHandler)
"""
# for checks
if not issubclass(handler_class, Handler):
raise exc.InterfaceError(f"Class {handler_class} " +
"does not implement Handler")
obj = handler_class()
# translate dashes to underscores
handler_class.Meta.label = re.sub('-', '_', obj._meta.label)
obj._meta.label = re.sub('-', '_', obj._meta.label)
interface = obj._meta.interface
LOG.debug("registering handler '%s' into handlers['%s']['%s']" %
(handler_class, interface, obj._meta.label))
if interface not in self.app.interface.list():
raise exc.InterfaceError(f"Handler interface '{interface}' doesn't exist.")
elif interface not in self.__handlers__.keys():
self.__handlers__[interface] = {}
if obj._meta.label in self.__handlers__[interface] and \
self.__handlers__[interface][obj._meta.label] != handler_class:
if force is True:
LOG.debug(
f"handlers['{interface}']['{obj._meta.label}'] already exists" +
", but `force==True`"
)
else:
raise exc.InterfaceError(
f"handlers['{interface}']['{obj._meta.label}'] already exists"
)
interface_class = self.app.interface.get(interface)
if not issubclass(handler_class, interface_class):
raise exc.InterfaceError(f"Handler {handler_class.__name__} " +
f"does not sub-class {interface_class.__name__}")
self.__handlers__[interface][obj._meta.label] = handler_class
def registered(self, interface: str, handler_label: str) -> bool:
"""
Check if a handler is registered.
Args:
interface (str): The interface of the handler (interface label)
handler_label (str): The label of the handler
Returns:
bool: ``True`` if the handler is registered, ``False`` otherwise
Example:
.. code-block:: python
app.handler.registered('log', 'colorlog')
"""
if interface in self.app.interface.list():
if interface in self.__handlers__.keys() and \
handler_label in self.__handlers__[interface]:
return True
return False
def setup(self, handler_class: Type[Handler]) -> Handler:
"""
Setup a handler class so that it can be used.
Args:
handler_class (class): An uninstantiated handler class.
Returns: None
Example:
.. code-block:: python
for controller in app.handler.list('controller'):
ch = app.handler.setup(controller)
"""
h = handler_class()
h._setup(self.app)
return h
def resolve(self,
interface: str,
handler_def: Union[str, Handler, Type[Handler]],
**kwargs: Any) -> Union[Handler, Optional[Handler]]:
"""
Resolves the actual handler, as it can be either a string identifying
the handler to load from ``self.__handlers__``, or it can be an
instantiated or non-instantiated handler class.
Args:
interface (str): The interface of the handler (ex: ``output``)
handler_def(str,instance,Handler): The loose references of the
handler, by label, instantiated object, or non-instantiated
class.
Keyword args:
raise_error (bool): Whether or not to raise an exception if unable
to resolve the handler.
meta_defaults (dict): Optional meta-data dictionary used as
defaults to pass when instantiating uninstantiated handlers.
Use ``App.Meta.meta_defaults`` by default.
setup (bool): Whether or not to call ``.setup()`` before return.
Default: ``False``
Returns:
instance: The instantiated handler object.
Example:
.. code-block:: python
# via label (str)
log = app.handler.resolve('log', 'colorlog')
# via uninstantiated handler class
log = app.handler.resolve('log', ColorLogHanddler)
# via instantiated handler instance
log = app.handler.resolve('log', ColorLogHandler())
"""
raise_error = kwargs.get('raise_error', True)
meta_defaults = kwargs.get('meta_defaults', None)
if meta_defaults is None:
meta_defaults = {}
if type(handler_def) is str:
_meta_label = f"{interface}.{handler_def}"
meta_defaults = self.app._meta.meta_defaults.get(_meta_label,
{})
elif hasattr(handler_def, 'Meta'):
_meta_label = f"{interface}.{handler_def.Meta.label}"
meta_defaults = self.app._meta.meta_defaults.get(_meta_label,
{})
setup = kwargs.get('setup', False)
han = None
if type(handler_def) is str:
han = self.get(interface, handler_def)(**meta_defaults) # type: ignore
elif hasattr(handler_def, '_meta'):
if not self.registered(interface, handler_def._meta.label): # type: ignore
self.register(handler_def.__class__) # type: ignore
han = handler_def
elif hasattr(handler_def, 'Meta'):
han = handler_def(**meta_defaults) # type: ignore
if not self.registered(interface, han._meta.label):
self.register(handler_def) # type: ignore
msg = f"Unable to resolve handler '{handler_def}' of interface '{interface}'"
if han is not None:
if setup is True:
han._setup(self.app)
return han
elif han is None and raise_error:
raise exc.FrameworkError(msg)
elif han is None:
LOG.debug(msg)
return None