def retrieve_influxDB_token()

in src/influxDBTokenPublisher.py [0:0]

• 22 lines of code
• 8 McCabe index (conditional complexity)

def retrieve_influxDB_token(args) -> str:
    """
    Retrieve the created RW token from InfluxDB.

    Parameters
    ----------
        args(Namespace): Parsed arguments

    Returns
    -------
        influxdb_rw_token(str): Parsed InfluxDB RW token.
    """

    token_json = ""
    dockerExecProcess = ""
    authListCommand = ['docker', 'exec', '-t', args.influxdb_container_name, 'influx', 'auth', 'list', '--json']
    if args.server_protocol == "https":
        authListCommand.append('--host')
        authListCommand.append('https://{}:{}'.format(args.influxdb_container_name, args.influxdb_port))

    if bool(strtobool(args.skip_tls_verify)):
        authListCommand.append('--skip-verify')

    logging.info("Running the following docker exec command to retrieve the InfluxDB token: " + str(authListCommand))

    dockerExecProcess = subprocess.run(authListCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    token_json = dockerExecProcess.stdout

    if dockerExecProcess.stderr:
        logging.error(dockerExecProcess.stderr)
    if(len(token_json) == 0):
        logging.error('Failed to retrieve InfluxDB RW token data from Docker! Retrieved token was: {}'.format(token_json))
        exit(1)
    influxdb_rw_token = next(d for d in json.loads(token_json) if d['description'] == 'greengrass_readwrite')['token']
    if(len(influxdb_rw_token) == 0):
        logging.error('Failed to parse InfluxDB RW token! Retrieved token was: {}'.format(influxdb_rw_token))
        exit(1)

    return influxdb_rw_token