witine/lookingglass
1
0
Fork 0
forked from mirrors/thatmattlove-hyperglass
Code Issues Pull requests Releases Activity

Implement plugins during execution workflow

Browse source
This commit is contained in:
thatmattlove 2021-09-13 02:37:32 -07:00
parent 3d97d118fb
commit 1adad7e46d
2 changed files with 7 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
hyperglass/execution/drivers/_common.py
View file

@ -2,7 +2,7 @@
# Standard Library
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Dict, Union, Sequence
from typing import TYPE_CHECKING, Union, Sequence
# Project
from hyperglass.log import log
@ -14,6 +14,7 @@ from ._construct import Construct
if TYPE_CHECKING:
# Project
from hyperglass.models.api import Query
from hyperglass.models.data import OutputDataModel
from hyperglass.compat._sshtunnel import SSHTunnelForwarder
from hyperglass.models.config.devices import Device
@ -36,7 +37,7 @@ class Connection(ABC):
"""Return a preconfigured sshtunnel.SSHTunnelForwarder instance."""
pass
async def parsed_response(self, output: Sequence[str]) -> Union[str, Sequence[Dict]]:
async def response(self, output: Sequence[str]) -> Union[OutputDataModel, str]:
"""Send output through common parsers."""
log.debug("Pre-parsed responses:\n{}", output)

7
hyperglass/execution/main.py
View file

@ -8,7 +8,7 @@ hyperglass-frr API calls, returns the output back to the front end.
# Standard Library
import signal
from typing import TYPE_CHECKING, Any, Dict, Union, Callable, Sequence
from typing import TYPE_CHECKING, Any, Dict, Union, Callable
# Project
from hyperglass.log import log
@ -18,6 +18,7 @@ from hyperglass.exceptions.public import DeviceTimeout, ResponseEmpty
if TYPE_CHECKING:
from hyperglass.models.api import Query
from .drivers import Connection
from hyperglass.models.data import OutputDataModel
# Local
from .drivers import AgentConnection, NetmikoConnection, ScrapliConnection
@ -44,7 +45,7 @@ def handle_timeout(**exc_args: Any) -> Callable:
return handler
async def execute(query: "Query") -> Union[str, Sequence[Dict]]:
async def execute(query: "Query") -> Union[OutputDataModel, str]:
"""Initiate query validation and execution."""
output = params.messages.general
@ -68,7 +69,7 @@ async def execute(query: "Query") -> Union[str, Sequence[Dict]]:
else:
response = await driver.collect()
output = await driver.parsed_response(response)
output = await driver.response(response)
if isinstance(output, str):
# If the output is a string (not structured) and is empty,