Source code for nipyapi.registry.configuration

# coding: utf-8

    Apache NiFi Registry REST API

    The REST API provides an interface to a registry with operations for saving, versioning, reading NiFi flows and components.

    OpenAPI spec version: 1.23.2
    Generated by:

from __future__ import absolute_import

import urllib3

import sys
import logging

from six import iteritems
from six.moves import http_client as httplib

[docs] def singleton(cls, *args, **kw): instances = {} def _singleton(): if cls not in instances: instances[cls] = cls(*args, **kw) return instances[cls] return _singleton
[docs] @singleton class Configuration(object): """ NOTE: This class is auto generated by the swagger code generator program. Ref: Do not edit the class manually. """ def __init__(self): """ Constructor """ # Default Base url = "http://localhost/nifi-registry-api" # Default api client self.api_client = None # Temp file folder for downloading files self.temp_folder_path = None # Authentication Settings self.force_basic_auth = False # dict to store API key(s) self.api_key = {} # dict to store API prefix (e.g. Bearer) self.api_key_prefix = {} # Username for HTTP basic authentication self.username = "" # Password for HTTP basic authentication self.password = "" # Logging Settings self.logger = {} self.logger["package_logger"] = logging.getLogger("swagger_client") self.logger["urllib3_logger"] = logging.getLogger("urllib3") # Log format self.logger_format = '%(asctime)s %(levelname)s %(message)s' # Log stream handler self.logger_stream_handler = None # Log file handler self.logger_file_handler = None # Debug file location self.logger_file = None # Debug switch self.debug = False # SSL/TLS verification # Set this to false to skip verifying SSL certificate when calling API from https server. self.verify_ssl = True # Set this to customize the certificate file to verify the peer. self.ssl_ca_cert = None # client certificate file self.cert_file = None # client key file self.key_file = None # Alternative TLS configuration: set this to an instance of ssl.SSLContext or urllib3.util.ssl_.SSLContext self.ssl_context = None # Proxy URL self.proxy = None # Safe chars for path_param self.safe_chars_for_path_param = '' @property def logger_file(self): """ Gets the logger_file. """ return self.__logger_file @logger_file.setter def logger_file(self, value): """ Sets the logger_file. If the logger_file is None, then add stream handler and remove file handler. Otherwise, add file handler and remove stream handler. :param value: The logger_file path. :type: str """ self.__logger_file = value if self.__logger_file: # If set logging file, # then add file handler and remove stream handler. self.logger_file_handler = logging.FileHandler(self.__logger_file) self.logger_file_handler.setFormatter(self.logger_formatter) for _, logger in iteritems(self.logger): logger.addHandler(self.logger_file_handler) if self.logger_stream_handler: logger.removeHandler(self.logger_stream_handler) else: # If not set logging file, # then add stream handler and remove file handler. self.logger_stream_handler = logging.StreamHandler() self.logger_stream_handler.setFormatter(self.logger_formatter) for _, logger in iteritems(self.logger): logger.addHandler(self.logger_stream_handler) if self.logger_file_handler: logger.removeHandler(self.logger_file_handler) @property def debug(self): """ Gets the debug status. """ return self.__debug @debug.setter def debug(self, value): """ Sets the debug status. :param value: The debug status, True or False. :type: bool """ self.__debug = value if self.__debug: # if debug status is True, turn on debug logging for _, logger in iteritems(self.logger): logger.setLevel(logging.DEBUG) # turn on httplib debug httplib.HTTPConnection.debuglevel = 1 else: # if debug status is False, turn off debug logging, # setting log level to default `logging.WARNING` for _, logger in iteritems(self.logger): logger.setLevel(logging.WARNING) # turn off httplib debug httplib.HTTPConnection.debuglevel = 0 @property def logger_format(self): """ Gets the logger_format. """ return self.__logger_format @logger_format.setter def logger_format(self, value): """ Sets the logger_format. The logger_formatter will be updated when sets logger_format. :param value: The format string. :type: str """ self.__logger_format = value self.logger_formatter = logging.Formatter(self.__logger_format) def get_api_key_with_prefix(self, identifier): """ Gets API key (with prefix if set). :param identifier: The identifier of apiKey. :return: The token for api key authentication. """ if self.api_key.get(identifier) and self.api_key_prefix.get(identifier): return self.api_key_prefix[identifier] + ' ' + self.api_key[identifier] elif self.api_key.get(identifier): return self.api_key[identifier] def get_basic_auth_token(self): """ Gets HTTP basic authentication header (string). :return: The token for basic HTTP authentication. """ return urllib3.util.make_headers(basic_auth=self.username + ':' + self.password)\ .get('authorization') def auth_settings(self): """ Gets Auth Settings dict for api client. :return: The Auth Settings information dict. """ return { 'tokenAuth': { 'type': 'api_key', 'in': 'header', 'key': 'Authorization', 'value': self.get_api_key_with_prefix('tokenAuth') }, 'basicAuth': { 'type': 'basic', 'in': 'header', 'key': 'Authorization', 'value': self.get_basic_auth_token() }, 'Authorization': { 'type': 'api_key', 'in': 'header', 'key': 'Authorization', 'value': self.get_api_key_with_prefix('Authorization') }, 'BasicAuth': { 'type': 'basic', 'in': 'header', 'key': 'Authorization', 'value': self.get_basic_auth_token() }, } def to_debug_report(self): """ Gets the essential information for debugging. :return: The report for debugging. """ return "Python SDK Debug Report:\n"\ "OS: {env}\n"\ "Python Version: {pyversion}\n"\ "Version of the API: 1.23.2\n"\ "SDK Package Version: 1.0.0".\ format(env=sys.platform, pyversion=sys.version)