magpie.register

Module Contents

Functions

_login_loop(login_url, cookies_file, data=None, message='Login response')
_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.
_phoenix_update_services(services_dict: magpie.typedefs.JSON) → bool
_phoenix_login(cookies)
_phoenix_login_check(cookies: magpie.typedefs.Str) → bool	Since Phoenix always return 200, even on invalid login, ‘hack’ check unauthorized access.
_phoenix_remove_services() → bool	Removes the Phoenix services using temporary cookies retrieved from login with defined PHOENIX constants.
_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]]
_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.
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.
_magpie_add_register_services_perms(services, statuses, curl_cookies, request_cookies, disable_getcapabilities)
_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.
_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 Services of loaded providers configuration using API requests.
_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 Services from providers section using pre-established database session.
_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.
get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Str, allow_missing: bool = False) → magpie.typedefs.MultiConfigs	Loads all matched configurations.
_expand_all(config: magpie.typedefs.JSON) → magpie.typedefs.JSON	Applies environment variable expansion recursively to all applicable fields of a configuration definition.
magpie_register_services_from_config(service_config_path: magpie.typedefs.Str, push_to_phoenix: 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.
_log_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) → None	Logs a message related to a ‘permission’ entry.
_use_request(cookies_or_session)
_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) → Tuple[Optional[int], bool]	Parses the resource field of a permission config entry and retrieves the final resource id. Creates missing
_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) → None	Applies the single permission entry retrieved from the permission configuration.
magpie_register_permissions_from_config(permissions_config: Union[magpie.typedefs.Str, magpie.typedefs.PermissionsConfig], magpie_url: Optional[magpie.typedefs.Str] = None, db_session: Optional[sqlalchemy.orm.session.Session] = None) → None	Applies permissions specified in configuration(s) defined as file, directory with files or literal configuration.
_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.
_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) → None	Processes a single permissions configuration.
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.

Attributes

LOGGER
LOGIN_ATTEMPT
LOGIN_TIMEOUT
CREATE_SERVICE_INTERVAL
GETCAPABILITIES_INTERVAL
GETCAPABILITIES_ATTEMPTS
SERVICES_MAGPIE
SERVICES_PHOENIX
SERVICES_PHOENIX_ALLOWED
CONFIG_KNOWN_EXTENSIONS

magpie.register.LOGGER

magpie.register.LOGIN_ATTEMPT = 5

magpie.register.LOGIN_TIMEOUT = 2

magpie.register.CREATE_SERVICE_INTERVAL = 2

magpie.register.GETCAPABILITIES_INTERVAL = 10

magpie.register.GETCAPABILITIES_ATTEMPTS = 12

magpie.register.SERVICES_MAGPIE = MAGPIE

magpie.register.SERVICES_PHOENIX = PHOENIX

magpie.register.SERVICES_PHOENIX_ALLOWED

exception magpie.register.RegistrationError

Bases: RuntimeError

Generic error during registration operation.

Initialize self. See help(type(self)) for accurate signature.

exception magpie.register.RegistrationValueError

Bases: RegistrationError, ValueError

Registration error caused by an invalid value precondition.

Initialize self. See help(type(self)) for accurate signature.

exception magpie.register.RegistrationLoginError

Bases: RegistrationError

Registration error caused by a failure to complete required login operation.

Initialize self. See help(type(self)) for accurate signature.

exception magpie.register.RegistrationConfigurationError

Bases: RegistrationValueError

Registration error caused by an invalid configuration entry or definition.

Initialize self. See help(type(self)) for accurate signature.

magpie.register._login_loop(login_url, cookies_file, data=None, message='Login response')

magpie.register._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

magpie.register._phoenix_update_services(services_dict: magpie.typedefs.JSON) → bool

magpie.register._phoenix_login(cookies)

magpie.register._phoenix_login_check(cookies: magpie.typedefs.Str) → bool

Since Phoenix always return 200, even on invalid login, ‘hack’ check unauthorized access.

Parameters

cookies – temporary cookies file storage used for login with _phoenix_login().

Returns

status indicating if login access was granted with defined credentials.

magpie.register._phoenix_remove_services() → bool

Removes the Phoenix services using temporary cookies retrieved from login with defined PHOENIX constants.

Returns

success status of the procedure.

magpie.register._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]]

magpie.register._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.

magpie.register.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 magpie.register.SERVICES_PHOENIX_ALLOWED.

Parameters

• services –

  An iterable of models.Service by default, or a dictionary of {svc-name: {<service-info>}} JSON objects containing each service’s information if services_ad_dicts is True.

  where <service-info> is defined as:

  {"public_url": <url>, "service_name": <name>, "service_type": <type>}
  

• services_as_dicts – indicate if services must be parsed as JSON definitions.

magpie.register._magpie_add_register_services_perms(services, statuses, curl_cookies, request_cookies, disable_getcapabilities)

magpie.register._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.

magpie.register._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 Services of loaded providers configuration using API requests.

See also

magpie_register_services_from_config()

Parameters

• services_dict – services configuration definition.

• push_to_phoenix – push registered Magpie services to Phoenix for synced configurations.

• username – login username to use to obtain permissions for services registration.

• password – login password to use to obtain permissions for services registration.

• provider – login provider to use to obtain permissions for services registration.

• force_update – override existing services matched by name

• disable_getcapabilities – do not execute ‘GetCapabilities’ validation for applicable services.

Returns

successful operation status

magpie.register._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 Services from providers section using pre-established database session.

See also

magpie_register_services_from_config()

magpie.register._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.

magpie.register.CONFIG_KNOWN_EXTENSIONS

magpie.register.get_all_configs(path_or_dict: Union[magpie.typedefs.Str, magpie.typedefs.CombinedConfig], section: magpie.typedefs.Str, allow_missing: bool = False) → magpie.typedefs.MultiConfigs

Loads all matched configurations.

Configurations are considered a valid match if they have one of the CONFIG_KNOWN_EXTENSIONS (if path) and that loaded (or passed) configurations contain the specified 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.

Parameters

• path_or_dict – directory path, file path or literal dictionary.

• section – section name that must be inside every matched configuration file to be loaded.

• 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.

magpie.register._expand_all(config: magpie.typedefs.JSON) → magpie.typedefs.JSON

Applies environment variable expansion recursively to all applicable fields of a configuration definition.

magpie.register.magpie_register_services_from_config(service_config_path: magpie.typedefs.Str, push_to_phoenix: 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.

Parameters

• service_config_path – where to look for providers configuration(s). Directory or file path.

• push_to_phoenix – whether to push loaded service definitions to remote Phoenix service.

• force_update – override service definitions that conflict by name with registered ones.

• 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.

• db_session – Use a pre-established database connection for registration. Otherwise, API requests are employed.

Returns

loaded service configurations.

magpie.register._log_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) → None

Logs a message related to a ‘permission’ entry.

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]

Parameters

• message – base message to log

• permission_index – index of the permission in the configuration list for traceability

• trail – trailing message appended after the base message

• detail – additional details appended after the trailing message after moving to another line.

• permission – permission name to log just before the trailing message.

• level – logging level (default: logging.WARN)

See also

magpie/config/permissions.cfg

magpie.register._use_request(cookies_or_session)

magpie.register._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) → 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)

magpie.register._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) → 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.

magpie.register.magpie_register_permissions_from_config(permissions_config: Union[magpie.typedefs.Str, magpie.typedefs.PermissionsConfig], magpie_url: Optional[magpie.typedefs.Str] = None, db_session: Optional[sqlalchemy.orm.session.Session] = None) → None

Applies permissions specified in configuration(s) defined as file, directory with files or literal configuration.

Parameters

• permissions_config – file/dir path to permissions config or JSON/YAML equivalent pre-loaded.

• magpie_url – URL to magpie instance (when using requests; default: magpie.url from this app’s config).

• db_session – db session to use instead of requests to directly create/remove permissions with config.

See also

magpie/config/permissions.cfg for specific parameters and operational details.

magpie.register._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 key to generate the mapping of unique items. First configuration entries have priority over later ones if keys are duplicated.

magpie.register._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) → None

Processes a single permissions configuration.

magpie.register.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.