def getTableProperties()

in visualizeConnectData/scripts/quicksight/deploy.py [0:0]
69 lines of code
23 McCabe index (conditional complexity)

def getTableProperties(catalog, database, tableName, s3Output):
    athenaClient = boto3.client('athena')
    
    sqeResp = athenaClient.start_query_execution(
        QueryString='DESCRIBE FORMATTED {0}'.format(tableName),
        QueryExecutionContext={
            'Database': database,
            'Catalog': catalog
        },
        ResultConfiguration={
            'OutputLocation': 's3://{0}'.format(s3Output),
        }
    )
    logger.info(sqeResp)
    queryExecutionId = sqeResp['QueryExecutionId']
    
    s3ResultLocation = None
    while True:
        time.sleep(sleepTime)
        
        gqeResp = athenaClient.get_query_execution(
            QueryExecutionId=queryExecutionId
        )
        logger.info(gqeResp)
        
        status = gqeResp['QueryExecution']['Status']['State']
        
        if (status == 'SUCCEEDED'):
            s3ResultLocation = gqeResp['QueryExecution']['ResultConfiguration']['OutputLocation']
            break
        
        if (status == 'FAILED'):
            raise RuntimeError('Athena get_query_execution failed')
    
    gqrResp = athenaClient.get_query_results(
        QueryExecutionId=queryExecutionId,
    )
    logger.info(gqrResp)
    resultSet = gqrResp['ResultSet']['Rows']
    
    tableInfo = []
    for row in resultSet:
        data = row['Data']
        if len(data) == 1:
            valueWhole = data[0]['VarCharValue']
            valueParts = re.split(r'\t+', valueWhole.rstrip('\t'))
            fieldName = valueParts[0].strip()
            dataType = valueParts[1].strip()
            dataType = dataType.lower()
            
            if (fieldName):
                if (fieldName == '# Detailed Table Information'):
                    break
                
                if (fieldName.startswith('#')):
                    continue
                
                quickSightDataType = None 
                if (dataType == 'string') or (dataType == 'char') or (dataType == 'varchar'):
                    quickSightDataType = 'STRING'
                elif (dataType == 'boolean'):
                    quickSightDataType = 'BOOLEAN'
                elif (dataType == 'tinyint') or (dataType == 'smallint') or (dataType == 'int') or (dataType == 'integer') or (dataType == 'bigint'):
                    quickSightDataType = 'INTEGER'
                elif (dataType == 'double') or (dataType == 'float') or (dataType == 'decimal'):
                    quickSightDataType = 'DECIMAL'
                elif (dataType == 'date') or (dataType == 'timestamp'):
                    quickSightDataType = 'DATETIME'
                else:
                    raise RuntimeError('Athena has an unsupported data type: {0}'.format(dataType))
                
                info = {
                    'name': fieldName.lower(),
                    'athenaDataType': dataType,
                    'quickSightDataType': quickSightDataType
                }
                
                tableInfo.append(info)
        else:
            raise RuntimeError('There should only be one data element')
    
    logger.info(tableInfo)
    return tableInfo