mirror/checktheroads-hyperglass
1
0
Fork 0
mirror of https://github.com/checktheroads/hyperglass synced 2024-05-11 05:55:08 +00:00
Code Releases Activity

Fix typing issues

Browse Source
This commit is contained in:
thatmattlove
2021-09-13 14:10:32 -07:00
parent 723048d1d1
commit f2cb15d0e2
4 changed files with 18 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hyperglass/execution/drivers/_common.py
View File

@ -37,7 +37,7 @@ class Connection(ABC):
"""Return a preconfigured sshtunnel.SSHTunnelForwarder instance."""
pass
async def response(self, output: Sequence[str]) -> Union[OutputDataModel, str]:
async def response(self, output: Sequence[str]) -> Union["OutputDataModel", str]:
"""Send output through common parsers."""
log.debug("Pre-parsed responses:\n{}", output)

2
hyperglass/execution/main.py
View File

@ -45,7 +45,7 @@ def handle_timeout(**exc_args: Any) -> Callable:
return handler
async def execute(query: "Query") -> Union[OutputDataModel, str]:
async def execute(query: "Query") -> Union["OutputDataModel", str]:
"""Initiate query validation and execution."""
output = params.messages.general

16
hyperglass/plugins/_base.py
View File

@ -54,7 +54,19 @@ class HyperglassPlugin(BaseModel, ABC):
super().__init__(name=name, **kwargs)
class DirectivePlugin(HyperglassPlugin):
"""Plugin associated with directives."""
class DirectivePlugin(BaseModel):
"""Plugin associated with directives.
Should always be subclassed with `HyperglassPlugin`.
"""
directives: Sequence[str] = ()
class DeviceTypePlugin(BaseModel):
"""Plugin associated with specific device types.
Should always be subclassed with `HyperglassPlugin`.
"""
device_types: Sequence[str] = ()

4
hyperglass/plugins/_output.py
View File

@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Union, Sequence
from hyperglass.log import log
# Local
from ._base import DirectivePlugin
from ._base import DirectivePlugin, DeviceTypePlugin, HyperglassPlugin
if TYPE_CHECKING:
# Project
@ -17,7 +17,7 @@ if TYPE_CHECKING:
OutputType = Union["OutputDataModel", Sequence[str]]
class OutputPlugin(DirectivePlugin):
class OutputPlugin(HyperglassPlugin, DirectivePlugin, DeviceTypePlugin):
"""Plugin to interact with device command output."""
def process(self, output: OutputType, device: "Device") -> OutputType: