artin/cement
1
0
Fork 0
mirror of https://github.com/datafolklabs/cement.git synced 2026-02-06 13:26:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Type Annotations: core.extension, core.handler

Browse Source 
- Resolves Issue #698
- Resolves Issue #700
- Related to PR #628
This commit is contained in:
BJ Dierkes 2024-06-23 22:44:18 -05:00
parent e502cab870
commit f75e810f7d
3 changed files with 64 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -29,6 +29,8 @@ Refactoring:
- `[core.controller]` [Issue #695](https://github.com/datafolklabs/cement/issues/695)
- `[core.deprecations]` [Issue #696](https://github.com/datafolklabs/cement/issues/696)
- `[core.exc]` [Issue #697](https://github.com/datafolklabs/cement/issues/697)
- `[core.extension]` [Issue #698](https://github.com/datafolklabs/cement/issues/698)
- `[core.handler]` [Issue #700](https://github.com/datafolklabs/cement/issues/700)
- `[core.interface]` [Issue #702](https://github.com/datafolklabs/cement/issues/702)
- `[core.meta]` [Issue #705](https://github.com/datafolklabs/cement/issues/705)
- `[utils.fs]` [Issue #688](https://github.com/datafolklabs/cement/issues/688)

32
cement/core/extension.py
View File

@ -1,7 +1,9 @@
"""Cement core extensions module."""
from __future__ import annotations
import sys
from abc import abstractmethod
from typing import Any, List, TYPE_CHECKING
from ..core import exc
from ..core.interface import Interface
from ..core.handler import Handler
@ -10,6 +12,10 @@ from ..utils.misc import minimal_logger
LOG = minimal_logger(__name__)
if TYPE_CHECKING:
from ..core.foundation import App # pragma: nocover
class ExtensionInterface(Interface):
"""
@ -19,15 +25,15 @@ class ExtensionInterface(Interface):
:class:`ExtensionHandler` base class as a starting point.
"""
class Meta:
class Meta(Interface.Meta):
"""Handler meta-data."""
#: The string identifier of the interface.
interface = 'extension'
interface: str = 'extension'
@abstractmethod
def load_extension(self, ext_module):
def load_extension(self, ext_module: str) -> None:
"""
Load an extension whose module is ``ext_module``. For example,
``cement.ext.ext_json``.
@ -39,7 +45,7 @@ class ExtensionInterface(Interface):
pass # pragma: no cover
@abstractmethod
def load_extensions(self, ext_list):
def load_extensions(self, ext_list: List[str]) -> None:
"""
Load all extensions from ``ext_list``.
@ -61,7 +67,7 @@ class ExtensionHandler(ExtensionInterface, Handler):
"""
class Meta:
class Meta(Handler.Meta):
"""
Handler meta-data (can be passed as keyword arguments to the parent
@ -69,14 +75,14 @@ class ExtensionHandler(ExtensionInterface, Handler):
"""
#: The string identifier of the handler.
label = 'cement'
label: str = 'cement'
def __init__(self, **kw):
def __init__(self, **kw: Any) -> None:
super().__init__(**kw)
self.app = None
self._loaded_extensions = []
self.app: App = None # type: ignore
self._loaded_extensions: List[str] = []
def get_loaded_extensions(self):
def get_loaded_extensions(self) -> List[str]:
"""
Get all loaded extensions.
@ -86,7 +92,7 @@ class ExtensionHandler(ExtensionInterface, Handler):
"""
return self._loaded_extensions
def list(self):
def list(self) -> List[str]:
"""
Synonymous with ``get_loaded_extensions()``.
@ -96,7 +102,7 @@ class ExtensionHandler(ExtensionInterface, Handler):
"""
return self._loaded_extensions
def load_extension(self, ext_module):
def load_extension(self, ext_module: str) -> None:
"""
Given an extension module name, load or in other-words ``import`` the
extension.
@ -132,7 +138,7 @@ class ExtensionHandler(ExtensionInterface, Handler):
except ImportError as e:
raise exc.FrameworkError(e.args[0])
def load_extensions(self, ext_list):
def load_extensions(self, ext_list: List[str]) -> None:
"""
Given a list of extension modules, iterate over the list and pass
individually to ``self.load_extension()``.

71
cement/core/handler.py
View File

@ -3,8 +3,10 @@ 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
@ -12,6 +14,10 @@ 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."""
@ -24,13 +30,13 @@ class Handler(ABC, MetaMixin):
"""
label = NotImplemented
label: str = NotImplemented
"""The string identifier of this handler."""
interface = NotImplemented
interface: str = NotImplemented
"""The interface that this class implements."""
config_section = None
config_section: str = None # type: ignore
"""
A config section to merge config_defaults with.
@ -39,14 +45,14 @@ class Handler(ABC, MetaMixin):
no section is set by the user/developer.
"""
config_defaults = None
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 = False
overridable: bool = False
"""
Whether or not handler can be overridden by
``App.Meta.handler_override_options``. Will be listed as an
@ -54,7 +60,7 @@ class Handler(ABC, MetaMixin):
``App.Meta.output_handler``, etc).
"""
def __init__(self, **kw):
def __init__(self, **kw: Any) -> None:
super(Handler, self).__init__(**kw)
try:
assert self._meta.label, \
@ -64,9 +70,9 @@ class Handler(ABC, MetaMixin):
except AssertionError as e:
raise exc.FrameworkError(e.args[0])
self.app = None
self.app: App = None # type: ignore
def _setup(self, app):
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
@ -88,11 +94,11 @@ class Handler(ABC, MetaMixin):
"into section '%s'" % 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.app.config.merge(dict_obj, override=False) # type: ignore
self._validate()
def _validate(self):
def _validate(self) -> None:
"""
Perform any validation to ensure proper data, meta-data, etc.
"""
@ -106,11 +112,15 @@ class HandlerManager(object):
"""
def __init__(self, app):
def __init__(self, app: App):
self.app = app
self.__handlers__ = {}
self.__handlers__: Dict[str, dict[str, Type[Handler]]] = {}
def get(self, interface, handler_label, fallback=None, **kwargs):
def get(self,
interface: str,
handler_label: str,
fallback: Optional[Type[Handler]] = None,
**kwargs: Any) -> Union[Handler, Type[Handler]]:
"""
Get a handler object.
@ -144,7 +154,7 @@ class HandlerManager(object):
"""
setup = kwargs.get('setup', False)
if interface not in self.app.interface.list():
if interface not in self.app.interface.list(): # type: ignore
raise exc.InterfaceError("Interface '%s' does not exist!" %
interface)
@ -160,7 +170,7 @@ class HandlerManager(object):
raise exc.InterfaceError("handlers['%s']['%s'] does not exist!" %
(interface, handler_label))
def list(self, interface):
def list(self, interface: str) -> List[Type[Handler]]:
"""
Return a list of handlers for a given ``interface``.
@ -181,7 +191,7 @@ class HandlerManager(object):
app.handler.list('log')
"""
if not self.app.interface.defined(interface):
if not self.app.interface.defined(interface): # type: ignore
raise exc.InterfaceError("Interface '%s' does not exist!" %
interface)
@ -190,7 +200,9 @@ class HandlerManager(object):
res.append(self.__handlers__[interface][label])
return res
def register(self, handler_class, force=False):
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
@ -243,7 +255,7 @@ class HandlerManager(object):
LOG.debug("registering handler '%s' into handlers['%s']['%s']" %
(handler_class, interface, obj._meta.label))
if interface not in self.app.interface.list():
if interface not in self.app.interface.list(): # type: ignore
raise exc.InterfaceError("Handler interface '%s' doesn't exist." %
interface)
elif interface not in self.__handlers__.keys():
@ -264,7 +276,7 @@ class HandlerManager(object):
(interface, obj._meta.label)
)
interface_class = self.app.interface.get(interface)
interface_class = self.app.interface.get(interface) # type: ignore
if not issubclass(handler_class, interface_class):
raise exc.InterfaceError("Handler %s " % handler_class.__name__ +
@ -273,7 +285,7 @@ class HandlerManager(object):
self.__handlers__[interface][obj._meta.label] = handler_class
def registered(self, interface, handler_label):
def registered(self, interface: str, handler_label: str) -> bool:
"""
Check if a handler is registered.
@ -291,14 +303,14 @@ class HandlerManager(object):
app.handler.registered('log', 'colorlog')
"""
if interface in self.app.interface.list():
if interface in self.app.interface.list(): # type: ignore
if interface in self.__handlers__.keys() and \
handler_label in self.__handlers__[interface]:
return True
return False
def setup(self, handler_class):
def setup(self, handler_class: Type[Handler]) -> Handler:
"""
Setup a handler class so that it can be used.
@ -319,7 +331,10 @@ class HandlerManager(object):
h._setup(self.app)
return h
def resolve(self, interface, handler_def, **kwargs):
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
@ -374,15 +389,15 @@ class HandlerManager(object):
han = None
if type(handler_def) is str:
han = self.get(interface, handler_def)(**meta_defaults)
han = self.get(interface, handler_def)(**meta_defaults) # type: ignore
elif hasattr(handler_def, '_meta'):
if not self.registered(interface, handler_def._meta.label):
self.register(handler_def.__class__)
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)
han = handler_def(**meta_defaults) # type: ignore
if not self.registered(interface, han._meta.label):
self.register(handler_def)
self.register(handler_def) # type: ignore
msg = "Unable to resolve handler '%s' of interface '%s'" % \
(handler_def, interface)