magpie.register

Module Contents

Functions

_login_loop(login_url, cookies_file, data=None, message=’Login response’)
_request_curl(url: Str, cookie_jar: Optional[Str] = None, cookies: Optional[Str] = None, form_params: Optional[Str] = None, msg: Optional[Str] = ‘Response’) → Tuple[int, int]	Executes a request using cURL.
_phoenix_update_services(services_dict: JSON) → bool
_phoenix_login(cookies)
_phoenix_login_check(cookies: 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[Str, Dict[Str, Any]], allowed_service_types: Optional[List[Str]] = None) → Tuple[bool, Dict[Str, int]]
_register_services(where: Optional[Str], services_dict: Dict[Str, Dict[Str, Str]], cookies: Str, message: Optional[Str] = ‘Register response’) → Tuple[bool, Dict[Str, int]]	Registers services on desired location using provided configurations and access cookies.
sync_services_phoenix(services: Union[Iterable[models.Service], 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[Str], services_dict: ConfigDict, request_cookies: AnyCookiesType) → Dict[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: ConfigDict, push_to_phoenix: bool, username: Str, password: Str, provider: Str, force_update: bool = False, disable_getcapabilities: bool = False) → bool	Registers magpie services using the provided services configuration.
_magpie_register_services_with_db_session(services_dict, db_session, push_to_phoenix=False, force_update=False, update_getcapabilities_permissions=False)
_load_config(path_or_dict: Union[Str, ConfigDict], section: Str, allow_missing: bool = False) → ConfigDict	Loads a file path or dictionary as YAML/JSON configuration.
get_all_configs(path_or_dict: Union[Str, ConfigDict], section: Str, allow_missing: bool = False) → List[ConfigDict]	Loads all configuration files specified by the path (if a directory), a single configuration (if a file) or directly
_expand_all(config: ConfigDict) → ConfigDict	Applies environment variable expansion recursively to all applicable fields of a configuration definition.
magpie_register_services_from_config(service_config_path: Str, push_to_phoenix: bool = False, force_update: bool = False, disable_getcapabilities: bool = False, db_session: Optional[Session] = None) → None	Registers Magpie services from one or many providers.cfg file.
_log_permission(message: Str, permission_index: int, trail: Str = ‘, skipping…’, detail: Optional[Str] = None, permission: Optional[Str] = None, level: Union[Str, int] = logging.WARN) → None	Logs a message related to a ‘permission’ entry.
_use_request(cookies_or_session)
_parse_resource_path(permission_config_entry: ConfigItem, entry_index: int, service_info: ConfigItem, cookies_or_session: CookiesOrSessionType = None, magpie_url: Optional[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: ConfigItem, entry_index: int, resource_id: int, cookies_or_session: CookiesOrSessionType, magpie_url: Str, users: ConfigDict, groups: ConfigDict) → None	Applies the single permission entry retrieved from the permission configuration.
magpie_register_permissions_from_config(permissions_config: Union[Str, ConfigDict], magpie_url: Optional[Str] = None, db_session: Optional[Session] = None) → None	Applies permissions specified in configuration(s) defined as file, directory with files or literal configuration.
_make_config_registry(config_entries: Optional[ConfigList], key: Str) → ConfigDict	Converts a list of configurations entries into a single mapping of configurations based on key.
_process_permissions(permissions: ConfigDict, magpie_url: Str, cookies_or_session: Session, users: Optional[ConfigList] = None, groups: Optional[ConfigList] = None) → None	Processes a single permissions configuration.
pseudo_random_string(length: int = 8, allow_chars: Str = string.ascii_letters + string.digits) → Str	Generate a string made of random characters.

magpie.register.ConfigItem

magpie.register.LOGGER

magpie.register.LOGIN_ATTEMPT = 10

magpie.register.LOGIN_TIMEOUT = 10

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: magpie.register.RegistrationError, ValueError

Registration error caused by an invalid value precondition.

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

exception magpie.register.RegistrationLoginError

Bases: magpie.register.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: magpie.register.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: Str, cookie_jar: Optional[Str] = None, cookies: Optional[Str] = None, form_params: Optional[Str] = None, msg: Optional[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: JSON) → bool

magpie.register._phoenix_login(cookies)

magpie.register._phoenix_login_check(cookies: 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[Str, Dict[Str, Any]], allowed_service_types: Optional[List[Str]] = None) → Tuple[bool, Dict[Str, int]]

magpie.register._register_services(where: Optional[Str], services_dict: Dict[Str, Dict[Str, Str]], cookies: Str, message: Optional[Str] = 'Register response') → Tuple[bool, Dict[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[models.Service], 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[Str], services_dict: ConfigDict, request_cookies: AnyCookiesType) → Dict[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: ConfigDict, push_to_phoenix: bool, username: Str, password: Str, provider: Str, force_update: bool = False, disable_getcapabilities: bool = False) → bool

Registers magpie services using the provided services configuration.

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, db_session, push_to_phoenix=False, force_update=False, update_getcapabilities_permissions=False)

magpie.register._load_config(path_or_dict: Union[Str, ConfigDict], section: Str, allow_missing: bool = False) → ConfigDict

Loads a file path or dictionary as YAML/JSON configuration.

magpie.register.get_all_configs(path_or_dict: Union[Str, ConfigDict], section: Str, allow_missing: bool = False) → List[ConfigDict]

Loads all configuration files specified by the path (if a directory), a single configuration (if a file) or directly returns the specified dictionary section (if a configuration dictionary).

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: ConfigDict) → ConfigDict

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

magpie.register.magpie_register_services_from_config(service_config_path: Str, push_to_phoenix: bool = False, force_update: bool = False, disable_getcapabilities: bool = False, db_session: Optional[Session] = None) → None

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.

magpie.register._log_permission(message: Str, permission_index: int, trail: Str = ', skipping...', detail: Optional[Str] = None, permission: Optional[Str] = None, level: Union[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: ConfigItem, entry_index: int, service_info: ConfigItem, cookies_or_session: CookiesOrSessionType = None, magpie_url: Optional[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: ConfigItem, entry_index: int, resource_id: int, cookies_or_session: CookiesOrSessionType, magpie_url: Str, users: ConfigDict, groups: ConfigDict) → 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[Str, ConfigDict], magpie_url: Optional[Str] = None, db_session: Optional[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._make_config_registry(config_entries: Optional[ConfigList], key: Str) → ConfigDict

Converts a list of configurations entries into a single mapping of configurations based on key.

First configuration entries have priority over later ones if keys are duplicated.

magpie.register._process_permissions(permissions: ConfigDict, magpie_url: Str, cookies_or_session: Session, users: Optional[ConfigList] = None, groups: Optional[ConfigList] = None) → None

Processes a single permissions configuration.

magpie.register.pseudo_random_string(length: int = 8, allow_chars: Str = string.ascii_letters + string.digits) → Str

Generate a string made of random characters.