Source code for nisystemlink.clients.core._http_configuration_manager

# -*- coding: utf-8 -*-

"""Implementation of HttpConfigurationManager."""

import json
import pathlib
import typing
from typing import Dict, Optional

from nisystemlink.clients import core
from nisystemlink.clients.core._internal._http_configuration_file import (
from nisystemlink.clients.core._internal._path_constants import PathConstants

[docs]class HttpConfigurationManager: """Factory for :class:`HttpConfiguration` objects.""" HTTP_MASTER_CONFIGURATION_ID = "SYSTEMLINK_MASTER" """The default ID of the SystemLink Server's configuration on systems registered using SystemLink Client.""" HTTP_LOCALHOST_CONFIGURATION_ID = "SYSTEMLINK_LOCALHOST" """The default ID of the SystemLink Server's configuration on the SystemLink Server itself.""" _configs = None
[docs] @classmethod def get_configuration( cls, id: Optional[str] = None, enable_fallbacks: Optional[bool] = True ) -> core.HttpConfiguration: """Get the requested or default configuration. Args: id: The ID of the configuration to find. enable_fallbacks: Whether or not to fallback to other known configurations, if ``id`` is unavailable. Returns: The configuration. Raises: ValueError: if ``id`` is None and ``enable_fallbacks`` is False. ApiException: if the specified (or default) configuration is not found. """ if id is not None: id = id.upper() else: if not enable_fallbacks: raise ValueError("id cannot be None if enable_fallbacks is False") fallback_configuration = cls._fallback() if fallback_configuration is None: raise core.ApiException("No SystemLink configurations available") return fallback_configuration if cls._configs is None: cls._configs = cls._read_configurations() config = cls._configs.get(id) if config is not None: return config if enable_fallbacks: fallback_configuration = cls._fallback() if fallback_configuration is None: raise core.ApiException( "Configuration with ID {!r} was not found, and no ".format(id) + "fallback configurations were available." ) return fallback_configuration raise core.ApiException("Configuration with ID {!r} was not found.".format(id))
@classmethod def _fallback(cls) -> Optional[core.HttpConfiguration]: """Attempt to acquire fallback HTTP configurations. Returns: The best available fallback configuration, or None if no such configurations are available. """ if cls._configs is None: cls._configs = cls._read_configurations() master_config = cls._configs.get(cls.HTTP_MASTER_CONFIGURATION_ID) if master_config is not None: return master_config localhost_config = cls._configs.get(cls.HTTP_LOCALHOST_CONFIGURATION_ID) if localhost_config is not None: return localhost_config return None @classmethod def _read_configurations(cls) -> Dict[str, core.HttpConfiguration]: """Discover and loads the HTTP configuration files. Returns: A dictionary mapping each loaded configuration ID to its corresponding :class:`HttpConfiguration`. Raises: ApiException: if multiple configurations with the same ID are simultaneously present. ApiException: if an OS or permission error prevents reading the directory that contains HTTP configurations. """ configurations = {} # type: Dict[str, core.HttpConfiguration] path = cls._http_configurations_directory() if not path.exists(): return configurations try: json_files = path.glob("*.json") except PermissionError as e: raise core.ApiException( "Not authorized to access HTTP configurations directory: " + str(e) ) except OSError as e: raise core.ApiException( "Error while accessing HTTP configurations directory: " + str(e) ) for json_file in json_files: try: config_file = cls._read_configuration_file(json_file) if config_file is None or is None: continue if in configurations: raise core.ApiException("Duplicate configuration IDs detected.") if not config_file.uri: continue cert_path = None # type: Optional[pathlib.Path] if config_file.cert_path: cert_path = typing.cast( pathlib.Path, PathConstants.application_data_directory / "Certificates" / config_file.cert_path, ) if not cert_path.exists(): cert_path = None configurations[] = core.HttpConfiguration( config_file.uri, config_file.api_key, cert_path=cert_path ) except PermissionError: pass except OSError: # The individual file is inaccessible or badly formatted, so just skip it pass except json.JSONDecodeError: pass except ValueError: pass return configurations @classmethod def _read_configuration_file( cls, path: pathlib.Path ) -> Optional[HttpConfigurationFile]: """Parse a single SystemLink HTTP configuration file. Args: path: The path of the file. Returns: The parsed file, or None if there is no valid configuration present at the given path. Raises: OSError: if an error occurs while accessing the file. PermissionError: if the current application does not have permission to access the configuration file. json.decoder.JSONDecodeError: if the file does not contain valid JSON. """ if not path.exists(): raise OSError("HTTP configuration file was not found: " + str(path)) with as f: config_dict = json.load(f) config = HttpConfigurationFile.from_json_dict(config_dict) if not return None = return config @classmethod def _http_configurations_directory(cls) -> pathlib.Path: """Get the platform-specific path to the HTTP Configurations directory. Returns: The path of the HTTP Configurations directory. """ return PathConstants.application_data_directory / "HttpConfigurations"