utils.py 文件源码

def _parse_job(self):
        try:
            setattr(
                self,
                'time_taken',
                dict(zip(
                    ('m', 's'),
                    divmod(
                        (
                            datetime.utcfromtimestamp(float(self.resp['statistics']['endTime']) / 1000) -
                            datetime.utcfromtimestamp(float(self.resp['statistics']['creationTime']) / 1000)
                        ).seconds,
                        60)
                ))
            )
        except KeyError:
            pass

        if self.job_type == 'load':
            try:
                setattr(self, 'size', humanize.naturalsize(int(self.resp['statistics']['load']['inputFileBytes'])))
            except (KeyError, TypeError):
                pass
        elif self.job_type == 'query':
            try:
                setattr(self, 'size', humanize.naturalsize(int(self.resp['statistics']['query']['totalBytesProcessed'])))
            except (KeyError, TypeError):
                pass

        if self.job_type == 'load':
            try:
                setattr(self, 'row_count', int(self.resp['statistics'][self.job_type]['outputRows']))
            except (KeyError, TypeError):
                pass
        elif self.job_type == 'query':
            try:
                setattr(self, 'row_count', int(self.resp['totalRows']))
            except (KeyError, TypeError):
                pass