Source code for cowbird.handlers.handler

import abc
import math
import os
from typing import Any, List
from typing_extensions import Literal

from cowbird.permissions_synchronizer import Permission
from cowbird.typedefs import JSON, SettingsType
from cowbird.utils import get_logger, get_ssl_verify, get_timeout

[docs]AnyHandlerParameter = Literal["priority", "url", "workspace_dir"]
[docs]HANDLER_PRIORITY_PARAM: AnyHandlerParameter = "priority"
[docs]HANDLER_URL_PARAM: AnyHandlerParameter = "url"
[docs]HANDLER_WORKSPACE_DIR_PARAM: AnyHandlerParameter = "workspace_dir"
[docs]LOGGER = get_logger(__name__)
[docs]class HandlerConfigurationException(Exception): """ Exception thrown when a handler cannot be instantiated because of a bad configuration. """
[docs]class Handler(abc.ABC):
[docs] __slots__ = ["settings", "name", "ssl_verify", "timeout", HANDLER_PRIORITY_PARAM, HANDLER_URL_PARAM, HANDLER_WORKSPACE_DIR_PARAM ]
""" Handler interface used to notify implemented handlers of users/permissions changes. .. todo:: At some point we will need a consistency function that goes through all Magpie users and make sure that handlers are up-to-date. """ @property @abc.abstractmethod
[docs] def required_params(self) -> List[AnyHandlerParameter]: raise NotImplementedError
def __init__(self, settings: SettingsType, name: str, **kwargs: Any) -> None: """ :param settings: Cowbird settings for convenience :param name: Handler name :param kwargs: The base class handle, but doesn't require the following variables: :param url: Location of the web service represented by the cowbird handler :param workspace_dir: Workspace directory :param priority: Relative priority between handlers while handling events. Lower value has higher priority, default value is last. """ if getattr(self, "required_params", None) is None: raise NotImplementedError("Handler 'required_params' must be overridden in inheriting class.") self.settings = settings = name self.priority = kwargs.get(HANDLER_PRIORITY_PARAM, math.inf) self.url = kwargs.get(HANDLER_URL_PARAM, None) self.workspace_dir = kwargs.get(HANDLER_WORKSPACE_DIR_PARAM, None) # Handlers making outbound requests should use these settings to avoid SSLError on test/dev setup self.ssl_verify = get_ssl_verify(self.settings) self.timeout = get_timeout(self.settings) for required_param in self.required_params: # pylint: disable=E1101,no-member if required_param not in HANDLER_PARAMETERS: raise HandlerConfigurationException(f"Invalid handler parameter : {required_param}") if getattr(self, required_param) is None: error_msg = f"{self.__class__.__name__} handler requires the following missing configuration " \ f"parameter : [{required_param}]" LOGGER.error(error_msg) raise HandlerConfigurationException(error_msg)
[docs] def json(self) -> JSON: return {"name":}
[docs] def _user_workspace_dir(self, user_name: str) -> str: return os.path.join(self.workspace_dir, user_name)
[docs] def user_created(self, user_name: str) -> None: raise NotImplementedError
[docs] def user_deleted(self, user_name: str) -> None: raise NotImplementedError
[docs] def permission_created(self, permission: Permission) -> None: raise NotImplementedError
[docs] def permission_deleted(self, permission: Permission) -> None: raise NotImplementedError
[docs] def resync(self) -> None: raise NotImplementedError