def _from_v1_resp()

in aliyun/log/getlogsresponse.py [0:0]

• 30 lines of code
• 4 McCabe index (conditional complexity)

    def _from_v1_resp(resp, header):
        """
        Support old api with best effort
        """
        meta_dict = {
            'count': len(resp),
            'progress': Util.h_v_t(header, 'x-log-progress'),
            'processedRows': int(Util.h_v_td(header, 'x-log-processed-rows', '0')),
            'elapsedMillisecond': int(Util.h_v_td(header, 'x-log-elapsed-millisecond', '0')),
            'hasSQL': (Util.h_v_td(header, 'x-log-has-sql', 'false') != 'false'),
            'whereQuery': Util.h_v_td(header, 'x-log-where-query', ''),
            'aggQuery': Util.h_v_td(header, 'x-log-agg-query', ''),
            'cpuSec': float(Util.h_v_td(header, 'x-log-cpu-sec', '0')),
            'cpuCores': int(Util.h_v_td(header, 'x-log-cpu-cores', '0')),
        }
        # parse query info
        query_info_str = Util.h_v_td(header, 'x-log-query-info', '')
        if query_info_str != '':
            query_info = json.loads(query_info_str)
            meta_dict['mode'] = Util.h_v_td(query_info, 'mode', 0)
            query_mode = GetLogsResponse.QueryMode(meta_dict['mode'])
            if query_mode in (GetLogsResponse.QueryMode.SCAN, GetLogsResponse.QueryMode.SCAN_SQL):
                meta_dict['scanBytes'] = Util.h_v_td(query_info, 'scanBytes', 0)
            if query_mode in (GetLogsResponse.QueryMode.PHRASE, GetLogsResponse.QueryMode.SCAN):
                scan_query_info = Util.h_v_td(query_info, 'phraseQueryInfo', dict())
                meta_dict['phraseQueryInfo'] = {
                    'scanAll': (Util.h_v_td(scan_query_info, 'scanAll', 'false') != 'false'), 
                    'beginOffset': int(Util.h_v_td(scan_query_info, 'beginOffset', '0')),
                    'endOffset': int(Util.h_v_td(scan_query_info, 'endOffset', '0'))
                }

        return GetLogsResponse({
            "meta": meta_dict,
            "data": resp
        }, header)