magpie.register =============== .. py:module:: magpie.register Attributes ---------- .. autoapisummary:: magpie.register.LOGGER magpie.register.LOGIN_ATTEMPT magpie.register.LOGIN_TIMEOUT magpie.register.CREATE_SERVICE_INTERVAL magpie.register.GETCAPABILITIES_INTERVAL magpie.register.GETCAPABILITIES_ATTEMPTS magpie.register.SERVICES_MAGPIE magpie.register.SERVICES_PHOENIX magpie.register.SERVICES_PHOENIX_ALLOWED magpie.register.CONFIG_KNOWN_EXTENSIONS Exceptions ---------- .. autoapisummary:: magpie.register.RegistrationError magpie.register.RegistrationValueError magpie.register.RegistrationLoginError magpie.register.RegistrationConfigurationError Functions --------- .. autoapisummary:: magpie.register._login_loop magpie.register._request_curl magpie.register._phoenix_update_services magpie.register._phoenix_login magpie.register._phoenix_login_check magpie.register._phoenix_remove_services magpie.register._phoenix_register_services magpie.register._register_services magpie.register.sync_services_phoenix magpie.register._magpie_add_register_services_perms magpie.register._magpie_update_services_conflict magpie.register._magpie_register_services_with_requests magpie.register._magpie_register_services_with_db_session magpie.register._load_config magpie.register.get_all_configs magpie.register._expand_all magpie.register.magpie_register_services_from_config magpie.register._handle_permission magpie.register._use_request magpie.register._parse_resource_path magpie.register._apply_permission_entry magpie.register.magpie_register_permissions_from_config magpie.register._resolve_config_registry magpie.register._process_permissions magpie.register.pseudo_random_string Module Contents --------------- .. py:data:: LOGGER .. py:data:: LOGIN_ATTEMPT :value: 5 .. py:data:: LOGIN_TIMEOUT :value: 2 .. py:data:: CREATE_SERVICE_INTERVAL :value: 2 .. py:data:: GETCAPABILITIES_INTERVAL :value: 10 .. py:data:: GETCAPABILITIES_ATTEMPTS :value: 12 .. py:data:: SERVICES_MAGPIE :value: 'MAGPIE' .. py:data:: SERVICES_PHOENIX :value: 'PHOENIX' .. py:data:: SERVICES_PHOENIX_ALLOWED .. py:exception:: RegistrationError Bases: :py:obj:`RuntimeError` Generic error during registration operation. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: RegistrationValueError Bases: :py:obj:`RegistrationError`, :py:obj:`ValueError` Registration error caused by an invalid value precondition. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: RegistrationLoginError Bases: :py:obj:`RegistrationError` Registration error caused by a failure to complete required login operation. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: RegistrationConfigurationError Bases: :py:obj:`RegistrationValueError` Registration error caused by an invalid configuration entry or definition. Initialize self. See help(type(self)) for accurate signature. .. py:function:: _login_loop(login_url, cookies_file, data=None, message='Login response') .. py:function:: _request_curl(url: magpie.typedefs.Str, cookie_jar: Optional[magpie.typedefs.Str] = None, cookies: Optional[magpie.typedefs.Str] = None, form_params: Optional[magpie.typedefs.Str] = None, msg: Optional[magpie.typedefs.Str] = 'Response') -> Tuple[int, int] Executes a request using cURL. :returns: tuple of the returned system command code and the response http code .. py:function:: _phoenix_update_services(services_dict: magpie.typedefs.JSON) -> bool .. py:function:: _phoenix_login(cookies_file: magpie.typedefs.Str) -> bool Performs Phoenix login using provided cookies. .. py:function:: _phoenix_login_check(cookies: magpie.typedefs.Str) -> bool Since Phoenix always return 200, even on invalid login, 'hack' check unauthorized access. :param cookies: temporary cookies file storage used for login with :func:`_phoenix_login`. :return: status indicating if login access was granted with defined credentials. .. py:function:: _phoenix_remove_services() -> bool Removes the Phoenix services using temporary cookies retrieved from login with defined `PHOENIX` constants. :returns: success status of the procedure. .. py:function:: _phoenix_register_services(services_dict: Dict[magpie.typedefs.Str, Dict[magpie.typedefs.Str, Any]], allowed_service_types: Optional[List[magpie.typedefs.Str]] = None) -> Tuple[bool, Dict[magpie.typedefs.Str, int]] .. py:function:: _register_services(where: Optional[magpie.typedefs.Str], services_dict: Dict[magpie.typedefs.Str, Dict[magpie.typedefs.Str, magpie.typedefs.Str]], cookies: magpie.typedefs.Str, message: Optional[magpie.typedefs.Str] = 'Register response') -> Tuple[bool, Dict[magpie.typedefs.Str, int]] Registers services on desired location using provided configurations and access cookies. :returns: tuple of overall success and individual http response of each service registration. .. py:function:: sync_services_phoenix(services: Union[Iterable[magpie.models.Service], magpie.typedefs.JSON], services_as_dicts: bool = False) -> bool Syncs Magpie services by pushing updates to Phoenix. Services must be one of types specified in :py:data:`magpie.register.SERVICES_PHOENIX_ALLOWED`. :param services: An iterable of :class:`models.Service` by default, or a dictionary of ``{svc-name: {}}`` JSON objects containing each service's information if :paramref:`services_ad_dicts` is ``True``. where ```` is defined as:: {"public_url": , "service_name": , "service_type": } :param services_as_dicts: indicate if services must be parsed as JSON definitions. .. py:function:: _magpie_add_register_services_perms(services: magpie.typedefs.ServicesSettings, statuses: Dict[magpie.typedefs.Str, int], curl_cookies: str, request_cookies: magpie.typedefs.AnyCookiesType, disable_getcapabilities: bool) -> None .. py:function:: _magpie_update_services_conflict(conflict_services: List[magpie.typedefs.Str], services_dict: magpie.typedefs.ServicesSettings, request_cookies: magpie.typedefs.AnyCookiesType) -> Dict[magpie.typedefs.Str, int] Resolve conflicting services by name during registration by updating them only if pointing to different URL. .. py:function:: _magpie_register_services_with_requests(services_dict: magpie.typedefs.ServicesSettings, push_to_phoenix: bool, username: magpie.typedefs.Str, password: magpie.typedefs.Str, provider: magpie.typedefs.Str, force_update: bool = False, disable_getcapabilities: bool = False) -> bool Registers :term:`Services` of loaded ``providers`` configuration using API requests. .. seealso:: :func:`magpie_register_services_from_config` :param services_dict: services configuration definition. :param push_to_phoenix: push registered Magpie services to Phoenix for synced configurations. :param username: login username to use to obtain permissions for services registration. :param password: login password to use to obtain permissions for services registration. :param provider: login provider to use to obtain permissions for services registration. :param force_update: override existing services matched by name :param disable_getcapabilities: do not execute 'GetCapabilities' validation for applicable services. :return: successful operation status .. py:function:: _magpie_register_services_with_db_session(services_dict: magpie.typedefs.ServicesSettings, db_session: sqlalchemy.orm.session.Session, push_to_phoenix: bool = False, force_update: bool = False, update_getcapabilities_permissions: bool = False) -> bool Registration procedure of :term:`Services` from ``providers`` section using pre-established database session. .. seealso:: :func:`magpie_register_services_from_config` .. py:function:: _load_config(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Str, allow_missing: bool = False) -> Union Loads a YAML/JSON file path or pre-loaded dictionary configuration. .. py:data:: CONFIG_KNOWN_EXTENSIONS .. 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:function:: _expand_all(config: magpie.typedefs.JSON) -> magpie.typedefs.JSON Applies environment variable expansion recursively to all applicable fields of a configuration definition. .. py:function:: magpie_register_services_from_config(service_config_path: magpie.typedefs.Str, push_to_phoenix: bool = False, skip_registration: bool = False, force_update: bool = False, disable_getcapabilities: bool = False, db_session: Optional[sqlalchemy.orm.session.Session] = None) -> magpie.typedefs.ServicesSettings Registers Magpie services from one or many `providers.cfg` file. Uses the provided DB session to directly update service definitions, or uses API request routes as admin. Optionally pushes updates to Phoenix. :param service_config_path: where to look for `providers` configuration(s). Directory or file path. :param push_to_phoenix: whether to push loaded service definitions to remote `Phoenix` service. :param skip_registration: Load, validate and combine :term:`Service` configurations, but don't register them. :param force_update: override service definitions that conflict by name with registered ones. :param disable_getcapabilities: Skip `GetCapabilities` request validation and permission update. By default, any service with `type` that allows `GetCapabilities` permissions will be tested to ensure it can be reached on the provided `url`. Once validated, this permission is applied to `anonymous` group to make its entrypoint accessible by anyone. Services that cannot have `GetCapabilities` permission are ignored regardless. :param db_session: Use a pre-established database connection for registration. Otherwise, API requests are employed. :returns: loaded service configurations. .. py:function:: _handle_permission(message: magpie.typedefs.Str, permission_index: int, trail: magpie.typedefs.Str = ', skipping...', detail: Optional[magpie.typedefs.Str] = None, permission: Optional[magpie.typedefs.Str] = None, level: Union[magpie.typedefs.Str, int] = logging.WARN, raise_errors: bool = False) -> None Logs a message related to a 'permission' entry and raises an error if required. Log message format is as follows (detail portion omitted if none provided):: {message} [permission: #{permission_index}] [{permission}]{trail} Detail: [{detail}] Such that the following logging entry is generated (omitting any additional logging formatters):: >> log_permission("test", 1, " skip test...", "just a test", "fake") test [permission: #1] [fake] skip test... Detail: [just a test] :param message: base message to log :param permission_index: index of the permission in the configuration list for traceability :param trail: trailing message appended after the base message :param detail: additional details appended after the trailing message after moving to another line. :param permission: permission name to log just before the trailing message. :param level: logging level (default: ``logging.WARN``) :param raise_errors: raises errors related to permissions, instead of just logging the info. .. seealso:: `magpie/config/permissions.cfg` .. py:function:: _use_request(cookies_or_session) .. py:function:: _parse_resource_path(permission_config_entry: magpie.typedefs.PermissionConfigItem, entry_index: int, service_info: magpie.typedefs.JSON, cookies_or_session: magpie.typedefs.CookiesOrSessionType = None, magpie_url: Optional[magpie.typedefs.Str] = None, raise_errors: bool = False) -> Tuple[Optional[int], bool] Parses the `resource` field of a permission config entry and retrieves the final resource id. Creates missing resources as necessary if they can be automatically resolved. If `cookies` are provided, uses requests to a running `Magpie` instance (with ``magpie_url``) to apply permission. If `session` to db is provided, uses direct db connection instead to apply permission. :returns: tuple of found id (if any, ``None`` otherwise), and success status of the parsing operation (error) .. py:function:: _apply_permission_entry(permission_config_entry: magpie.typedefs.PermissionConfigItem, entry_index: int, resource_id: int, cookies_or_session: magpie.typedefs.CookiesOrSessionType, magpie_url: magpie.typedefs.Str, users: magpie.typedefs.UsersSettings, groups: magpie.typedefs.GroupsSettings, raise_errors: bool = False) -> None Applies the single permission entry retrieved from the permission configuration. Assumes that permissions fields where pre-validated. Permission is applied for the user/group/resource using request or db session accordingly to arguments. .. py:function:: magpie_register_permissions_from_config(permissions_config: Union[magpie.typedefs.Str, magpie.typedefs.PermissionsConfig], settings: Optional[magpie.typedefs.AnySettingsContainer] = None, db_session: Optional[sqlalchemy.orm.session.Session] = None, raise_errors: bool = False) -> None Applies `permissions` specified in configuration(s) defined as file, directory with files or literal configuration. :param permissions_config: file/dir path to `permissions` config or JSON/YAML equivalent pre-loaded. :param settings: Magpie settings to resolve an instance session when using requests instead of DB session. Will look for ``magpie.url``, ``magpie.admin_user`` and ``magpie.admin_password`` by default, or any corresponding environment variable resolution if omitted in the settings. :param db_session: db session to use instead of requests to directly create/remove permissions with config. :param raise_errors: raises errors related to permissions, instead of just logging the info. .. seealso:: `magpie/config/permissions.cfg` for specific parameters and operational details. .. py:function:: _resolve_config_registry(config_files: Optional[magpie.typedefs.MultiConfigs], key: magpie.typedefs.Str) -> magpie.typedefs.AnyResolvedSettings Converts a list of configurations entries from multiple files into a single resolved mapping. Resolution is accomplished against :paramref:`key` to generate the mapping of unique items. First configuration entries have priority over later ones if keys are duplicated. .. py:function:: _process_permissions(permissions: magpie.typedefs.PermissionsConfig, magpie_url: magpie.typedefs.Str, cookies_or_session: sqlalchemy.orm.session.Session, users: Optional[magpie.typedefs.UsersSettings] = None, groups: Optional[magpie.typedefs.GroupsSettings] = None, settings: Optional[magpie.typedefs.AnySettingsContainer] = None, raise_errors: bool = False) -> None Processes a single `permissions` configuration. .. py:function:: pseudo_random_string(length: int = 8, allow_chars: magpie.typedefs.Str = string.ascii_letters + string.digits) -> magpie.typedefs.Str Generate a string made of random characters.