forked from Support/sp-hydra-veil-gui
74 lines
2.4 KiB
Python
Executable file
74 lines
2.4 KiB
Python
Executable file
import inspect
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from core.controllers.ConfigurationController import ConfigurationController
|
|
from core.models.Configuration import ConnectionChoice
|
|
from core.models.Result import ResultError
|
|
from core.services.prepare_tickets import ticket_tracker
|
|
|
|
|
|
DEFAULT_ENUM_MAP = {
|
|
ResultError.SUBSCRIPTION: (
|
|
ticket_tracker.wipe_one_ticket_sub,
|
|
{},
|
|
),
|
|
ResultError.CONNECTION: (
|
|
ConfigurationController.set_connection_enum,
|
|
{'connection_enum': ConnectionChoice.SYSTEM},
|
|
),
|
|
ResultError.INVALID_API_REPLY: None,
|
|
}
|
|
|
|
|
|
def has_function(enum_map: dict[Enum, Any], enum_key: Enum) -> bool:
|
|
action = enum_map.get(enum_key)
|
|
if action is None:
|
|
return False
|
|
function, _ = _normalize_action(action)
|
|
return callable(function)
|
|
|
|
|
|
def error_dispatch(enum_map: dict[Enum, Any], enum_key: Enum, **kwargs: Any) -> Any:
|
|
if enum_key not in enum_map:
|
|
return None
|
|
action = enum_map[enum_key]
|
|
if action is None:
|
|
return None
|
|
function, default_kwargs = _normalize_action(action)
|
|
call_kwargs = dict(default_kwargs)
|
|
call_kwargs.update(kwargs)
|
|
return function(**_filter_kwargs(function, call_kwargs))
|
|
|
|
|
|
def has_result_action(result, enum_map: dict[Enum, Any] | None = None) -> bool:
|
|
return has_function(enum_map or DEFAULT_ENUM_MAP, result.error_type)
|
|
|
|
|
|
def dispatch_result_action(result, enum_map: dict[Enum, Any] | None = None, **kwargs: Any) -> Any:
|
|
return error_dispatch(enum_map or DEFAULT_ENUM_MAP, result.error_type, **kwargs)
|
|
|
|
|
|
def _normalize_action(action: Any) -> tuple[Any, dict[str, Any]]:
|
|
if isinstance(action, tuple):
|
|
function = action[0]
|
|
if len(action) > 1 and isinstance(action[1], dict):
|
|
return function, action[1]
|
|
return function, {}
|
|
return action, {}
|
|
|
|
|
|
def _filter_kwargs(function: Any, kwargs: dict[str, Any]) -> dict[str, Any]:
|
|
signature = inspect.signature(function)
|
|
parameters = signature.parameters
|
|
if any(param.kind == inspect.Parameter.VAR_KEYWORD for param in parameters.values()):
|
|
return kwargs
|
|
accepted = {
|
|
name
|
|
for name, param in parameters.items()
|
|
if param.kind in (
|
|
inspect.Parameter.POSITIONAL_OR_KEYWORD,
|
|
inspect.Parameter.KEYWORD_ONLY,
|
|
)
|
|
}
|
|
return {key: value for key, value in kwargs.items() if key in accepted}
|