magpie.api.webhooks =================== .. py:module:: magpie.api.webhooks Attributes ---------- .. autoapisummary:: magpie.api.webhooks.CONTENT_TYPE_JSON magpie.api.webhooks.FORMAT_TYPE_MAPPING magpie.api.webhooks.WEBHOOK_KEYS_REQUIRED magpie.api.webhooks.WEBHOOK_KEYS_OPTIONAL magpie.api.webhooks.WEBHOOK_KEYS magpie.api.webhooks.WEBHOOK_TEMPLATE_PARAMS magpie.api.webhooks.WEBHOOK_HTTP_METHODS magpie.api.webhooks.LOGGER magpie.api.webhooks.WebhookActionNames Classes ------- .. autoapisummary:: magpie.api.webhooks.ExtendedEnum magpie.api.webhooks.WebhookAction Functions --------- .. autoapisummary:: magpie.api.webhooks.format_group magpie.api.webhooks.format_resource magpie.api.webhooks.format_service magpie.api.webhooks.format_user magpie.api.webhooks.get_constant magpie.api.webhooks.get_db_session_from_config_ini magpie.api.webhooks.get_all_configs magpie.api.webhooks.get_logger magpie.api.webhooks.get_settings magpie.api.webhooks.raise_log magpie.api.webhooks.signature_with_args magpie.api.webhooks.get_permission_update_params magpie.api.webhooks.process_webhook_requests magpie.api.webhooks.generate_callback_url magpie.api.webhooks.replace_template magpie.api.webhooks.send_webhook_request magpie.api.webhooks.webhook_update_error_status magpie.api.webhooks.setup_webhooks Module Contents --------------- .. py:function:: format_group(group: magpie.models.Group, basic_info: bool = False, public_info: bool = False, dotted: bool = False, db_session: Optional[sqlalchemy.orm.session.Session] = None) -> magpie.typedefs.JSON Obtains the JSON formatted :term:`Group` definition according to field selection flags. :param group: Group for which to provide details. :param basic_info: If ``True``, return only sufficient details to identify the group (useful for routes that refer to a group, but that are not requesting it specifically), or return full details (for routes that specifically request its information, e.g.: ``GET /groups/{grp}``). :param public_info: Indicate if the returned details are intended for public information (``True``) or admin-only (``False``). Only higher level users should be provided additional details to avoid leaking potentially sensitive parameters. :param dotted: Employ a dot (``.``) instead of underscore (``_``) to separate :term:`Group` from its basic information. :param db_session: Database connection to retrieve additional details (required when ``public_info=False``). .. py:function:: format_resource(resource: magpie.models.Resource, permissions: Optional[Collection[magpie.typedefs.AnyPermissionType]] = None, permission_type: Optional[magpie.permissions.PermissionType] = None, basic_info: bool = False, dotted: bool = False) -> magpie.typedefs.JSON Formats a :term:`Resource` information into JSON. :param resource: :term:`Resource` to be formatted. :param permissions: Permissions to list along with the :paramref:`resource`. By default, these are the applicable permissions for that corresponding resource type. :param permission_type: Override indication of provenance to apply to :paramref:`permissions`. Only applicable when they are provided. :param basic_info: If ``True``, return only sufficient details to identify the resource, without any additional :paramref:`permissions` detail, nor hierarchical :paramref:`resource` information is returned. :param dotted: Employ a dot (``.``) instead of underscore (``_``) to separate :term:`Resource` from its basic information. .. seealso:: :func:`magpie.api.management.service.service_formats.format_service` .. py:function:: format_service(service: magpie.models.Service, permissions: Optional[List[magpie.permissions.PermissionSet]] = None, permission_type: Optional[magpie.permissions.PermissionType] = None, show_private_url: bool = False, show_public_url: bool = True, show_resources_allowed: bool = False, show_configuration: bool = False, basic_info: bool = False, dotted: bool = False) -> magpie.typedefs.JSON Formats a :term:`Service` information into JSON. .. note:: Automatically finds :paramref:`permissions` of the :paramref:`service` if not specified. To preserve `empty` permissions such as during listing of `user`/`group` resource permissions, an empty ``list`` should be specified. :param service: :term:`Service` to be formatted. :param permissions: Permissions to list along with the :paramref:`resource`. By default, these are the applicable permissions for that corresponding resource type. :param permission_type: Override indication of provenance to apply to :paramref:`permissions`. Only applicable when they are provided. :param show_private_url: Display the protected and private URL employed at service registration. :param show_public_url: Display the generated public URL from configured :ref:`config_twitcher`. :param show_resources_allowed: Display children resource details. :param show_configuration: Display the applicable configuration of the :term:`Service` if it supports it. :param basic_info: If ``True``, return only sufficient details to identify the service, without any additional details about :paramref:`permissions`, children resources or configuration information is returned. :param dotted: Employ a dot (``.``) instead of underscore (``_``) to separate :term:`Service` from its basic information. .. seealso:: :func:`magpie.api.management.resource.resource_formats.format_resource` .. py:function:: format_user(user: magpie.models.AnyUser, group_names: List[magpie.typedefs.Str] = None, basic_info: bool = False, dotted: bool = False) -> magpie.typedefs.JSON Formats a :term:`User` information into JSON. :param user: :term:`User` to be formatted. :param group_names: Override of group name listing to provide as user memberships. Retrieve the complete group membership listing from the :paramref:`user` if not provided. :param basic_info: If ``True``, return only sufficient details to identify the :term:`User`, without any additional group details. :param dotted: Employ a dot (``.``) instead of underscore (``_``) to separate :term:`User` from its basic information. .. seealso:: :func:`magpie.api.management.group.group_formats.format_group` .. py:function:: get_constant(constant_name: magpie.typedefs.Str, settings_container: Optional[magpie.typedefs.AnySettingsContainer] = None, settings_name: Optional[magpie.typedefs.Str] = None, default_value: Optional[magpie.typedefs.SettingValue] = None, raise_not_set: bool = True, raise_missing: bool = True, print_missing: bool = False, empty_missing: bool = False) -> magpie.typedefs.SettingValue Search in order for matched value of :paramref:`constant_name`: 1. search in :py:data:`MAGPIE_CONSTANTS` 2. search in settings if specified 3. search alternative setting names (see below) 4. search in :mod:`magpie.constants` definitions 5. search in environment variables Parameter :paramref:`constant_name` is expected to have the format ``MAGPIE_[VARIABLE_NAME]`` although any value can be passed to retrieve generic settings from all above-mentioned search locations. If :paramref:`settings_name` is provided as alternative name, it is used as is to search for results if :paramref:`constant_name` was not found. Otherwise, ``magpie.[variable_name]`` is used for additional search when the format ``MAGPIE_[VARIABLE_NAME]`` was used for :paramref:`constant_name` (i.e.: ``MAGPIE_ADMIN_USER`` will also search for ``magpie.admin_user`` and so on for corresponding constants). :param constant_name: key to search for a value :param settings_container: WSGI application settings container (if not provided, uses found one in current thread) :param settings_name: alternative name for `settings` if specified :param default_value: default value to be returned if not found anywhere, and exception raises are disabled. :param raise_not_set: raise an exception if the found key is ``None``, search until last case if others are ``None`` :param raise_missing: raise exception if key is not found anywhere :param print_missing: print message if key is not found anywhere, return ``None`` :param empty_missing: consider an empty value for an existing key as if it was missing (i.e.: as if not set). :returns: found value or `default_value` :raises ValueError: if resulting value is invalid based on options (by default raise missing/empty/``None`` value) :raises LookupError: if no appropriate value could be found from all search locations (according to options) .. py:function:: get_db_session_from_config_ini(config_ini_path, ini_main_section_name='app:magpie_app', settings_override=None) .. py:function:: get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Literal[groups], allow_missing: bool = False) -> magpie.typedefs.GroupsConfig get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Literal[users], allow_missing: bool = False) -> magpie.typedefs.UsersConfig get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Literal[permissions], allow_missing: bool = False) -> magpie.typedefs.PermissionsConfig get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Literal[services], allow_missing: bool = False) -> magpie.typedefs.ServicesConfig get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Literal[webhooks], allow_missing: bool = False) -> magpie.typedefs.WebhooksConfig Loads all matched configurations. Configurations are considered a valid match if they have one of the :py:data:`CONFIG_KNOWN_EXTENSIONS` (if path) and that loaded (or passed) configurations contain the specified :paramref:`section` name. If the input is a directory path, loads any number of files contained in it that fulfill matching conditions. If it is a path pointing to a single valid configuration file, loads it by itself. If a dictionary is passed, returns it directly if it fulfills validation. :param path_or_dict: directory path, file path or literal dictionary. :param section: section name that must be inside every matched configuration file to be loaded. :param allow_missing: allow to have no valid configuration after all are resolved, otherwise raises. :raises RegistrationError: when no valid configuration can be found and empty one is not allowed. :returns: - list of configurations loaded if input was a directory path - list of single configuration if input was a file path - list of single configuration if input was a JSON dict - empty list if none of the other cases where matched .. note:: Order of file loading will be resolved by alphabetically sorted filename if specifying a directory path. .. py:data:: CONTENT_TYPE_JSON :value: 'application/json' .. py:data:: FORMAT_TYPE_MAPPING .. py:class:: ExtendedEnum Bases: :py:obj:`enum.Enum` Utility :class:`enum.Enum` methods. Create an extended enum with these utilities as follows:: class CustomEnum(ExtendedEnum): ItemA = "A" ItemB = "B" .. py:method:: names() -> List[magpie.typedefs.Str] :classmethod: Returns the member names assigned to corresponding enum elements. .. py:method:: values() -> List[magpie.typedefs.AnyKey] :classmethod: Returns the literal values assigned to corresponding enum elements. .. py:method:: get(key_or_value: magpie.typedefs.AnyKey, default: Optional[Any] = None) -> Optional[_TC] :classmethod: Finds an enum entry by defined name or its value. Returns the entry directly if it is already a valid enum. .. py:method:: titles() -> List[magpie.typedefs.Str] :classmethod: Returns the title representation of all enum elements. .. py:property:: title :type: magpie.typedefs.Str Returns the title representation of the enum element. Title use the original enum element name with capitalization considering underscores for separate words. .. py:function:: get_logger(name: magpie.typedefs.Str, level: Optional[int] = None, force_stdout: bool = None, message_format: Optional[magpie.typedefs.Str] = None, datetime_format: Optional[magpie.typedefs.Str] = None) -> logging.Logger Immediately sets the logger level to avoid duplicate log outputs from the `root logger` and `this logger` when `level` is ``logging.NOTSET``. .. py:function:: get_settings(container: Optional[magpie.typedefs.AnySettingsContainer], app: bool = False) -> magpie.typedefs.SettingsType Retrieve application settings from a supported container. :param container: supported container with a handle to application settings. :param app: allow retrieving from current thread registry if no container was defined. :return: found application settings dictionary. :raise TypeError: when no application settings could be found or unsupported container. .. py:function:: raise_log(msg: magpie.typedefs.Str, exception: Type[Exception] = Exception, logger: Optional[logging.Logger] = None, level: int = logging.ERROR) -> NoReturn Logs the provided message to the logger and raises the corresponding exception afterwards. :raises exception: whichever exception provided is raised systematically after logging. .. py:function:: signature_with_args(func: Callable[[Ellipsis], Any], *args: Any, **kwargs: Any) -> magpie.typedefs.Str Returns a visual representation of a function signature with is arguments. .. code-block:: python def function(a, b, c=1): ... signature_with_args(function, 1, 2, c=3) # returns: .. py:data:: WEBHOOK_KEYS_REQUIRED .. py:data:: WEBHOOK_KEYS_OPTIONAL .. py:data:: WEBHOOK_KEYS .. py:data:: WEBHOOK_TEMPLATE_PARAMS :value: ['group.name', 'group.id', 'user.name', 'user.id', 'user.email', 'user.status', 'resource.id',... .. py:data:: WEBHOOK_HTTP_METHODS :value: ['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE'] .. py:data:: LOGGER .. py:class:: WebhookAction Bases: :py:obj:`magpie.utils.ExtendedEnum` Supported :term:`Webhook` actions. .. py:attribute:: CREATE_USER :value: 'create_user' Triggered when a new :term:`User` gets successfully created. .. seealso:: :ref:`webhook_user_create` .. py:attribute:: DELETE_USER :value: 'delete_user' Triggered when an existing :term:`User` gets successfully deleted. .. seealso:: :ref:`webhook_user_delete` .. py:attribute:: UPDATE_USER_STATUS :value: 'update_user_status' Triggered when an existing :term:`User` status gets successfully updated. .. seealso:: :ref:`webhook_user_update_status` .. py:attribute:: CREATE_USER_PERMISSION :value: 'create_user_permission' Triggered when a :term:`Permission` onto a :term:`Service` or :term:`Resource` gets created for a :term:`User`. .. seealso:: :ref:`webhook_permission_updates` .. py:attribute:: DELETE_USER_PERMISSION :value: 'delete_user_permission' Triggered when a :term:`Permission` onto a :term:`Service` or :term:`Resource` gets deleted for a :term:`User`. .. seealso:: :ref:`webhook_permission_updates` .. py:attribute:: CREATE_GROUP_PERMISSION :value: 'create_group_permission' Triggered when a :term:`Permission` onto a :term:`Service` or :term:`Resource` gets created for a :term:`Group`. .. seealso:: :ref:`webhook_permission_updates` .. py:attribute:: DELETE_GROUP_PERMISSION :value: 'delete_group_permission' Triggered when a :term:`Permission` onto a :term:`Service` or :term:`Resource` gets deleted for a :term:`Group`. .. seealso:: :ref:`webhook_permission_updates` .. py:data:: WebhookActionNames .. py:function:: get_permission_update_params(target: Union[magpie.models.User, magpie.models.Group], resource: magpie.typedefs.ServiceOrResourceType, permission: magpie.permissions.PermissionSet, db_session: sqlalchemy.orm.session.Session) -> magpie.typedefs.WebhookTemplateParameters Generates the :term:`Webhook` parameters based on provided references. .. py:function:: process_webhook_requests(action: WebhookAction, params: magpie.typedefs.WebhookTemplateParameters, update_user_status_on_error: bool = False, settings: Optional[magpie.typedefs.AnySettingsContainer] = None) -> None Checks the config for any webhooks that correspond to the input action, and prepares corresponding requests. :param action: tag identifying which webhooks to use in the config :param params: Dictionary containing the required parameters and associated values for the request following the event action. Parameters will replace *templates* found in the ``payload`` definition of the webhook. :param update_user_status_on_error: update the user status or not in case of a webhook error. :param settings: application settings where webhooks configuration can be retrieved. .. py:function:: generate_callback_url(operation: magpie.models.TokenOperation, db_session: sqlalchemy.orm.session.Session, user: Optional[magpie.models.AnyUser] = None, group: Optional[magpie.models.Group] = None) -> magpie.typedefs.Str Generates a callback URL using `Magpie` temporary tokens for use by the webhook implementation. :param operation: targeted operation that employs the callback URL for reference. :param db_session: database session to store the generated temporary token. :param user: user reference associated to the operation as applicable. :param group: group reference associated to the operation as applicable. :return: generated callback URL. .. py:function:: replace_template(params: magpie.typedefs.WebhookTemplateParameters, payload: magpie.typedefs.WebhookPayload, force_str: bool = False) -> magpie.typedefs.WebhookPayload Replace each template parameter from the payload by its corresponding value. :param params: the values of the template parameters :param payload: structure containing the data to be processed by the template replacement :param force_str: enforce string conversion of applicable fields where non-string values are detected. :return: structure containing the data with the replaced template parameters .. py:function:: send_webhook_request(webhook_config: magpie.typedefs.WebhookConfigItem, params: magpie.typedefs.WebhookTemplateParameters, update_user_status_on_error: bool = False) -> None Sends a single webhook request using the input config. :param webhook_config: dictionary containing the config data of a single webhook :param params: dictionary containing the required parameters for the request, they will replace templates found in the payload :param update_user_status_on_error: update the user status or not in case of a webhook error .. py:function:: webhook_update_error_status(user_name: magpie.typedefs.Str) -> None Updates the user's status to indicate an error occurred with the webhook requests. .. py:function:: setup_webhooks(config_path: Optional[magpie.typedefs.Str], settings: magpie.typedefs.SettingsType) -> None Prepares and validates :term:`Webhook` settings for the application based on definitions in configuration file(s). Following execution, all validated :term:`Webhook` configurations will have every parameters defined in :py:data:`WEBHOOK_KEYS`, whether optional or mandatory. Required parameters in :py:data:`WEBHOOK_KEYS_REQUIRED` are explicitly validated for defined value and raise if missing. Parameters from :py:data:`WEBHOOK_KEYS_OPTIONAL` are defaulted to ``None`` if missing. Any :term:`Webhook` failing validation will raise the whole configuration and not apply any changes to the :paramref:`settings`. Format validation is applied to some specific parameters to anticipate and raise definitions guaranteed to be erroneous to avoid waiting until runtime for them to fail upon their trigger event. .. seealso:: Documentation in :ref:`config_webhook`. :param config_path: a single file or directory path where configuration file(s) with ``webhook`` section. :param settings: modified settings in-place with added valid webhooks.