rudra/data_store/bigquery/pypi_bigquery.py
def process(self):
        """Process Pypi Bigquery response data."""
        start = time.monotonic()
        logger.info("Running Bigquery for pypi synchronously")
        self.big_query_instance.run_query_sync()

        logger.info("fetching bigquery result.")
        for content in self.big_query_instance.get_result():
            self.big_query_content.append(content)
            logger.info("collected manifests: {}".format(len(self.big_query_content)))
        logger.info("Succefully retrieved data from Bigquery, time:{}".format(
            time.monotonic() - start))
        base_url_pypi = 'https://pypi.org/pypi/{pkg}/json'
        logger.info("Starting package cleaning")
        start_process_time = time.monotonic()
        for idx, obj in enumerate(self.big_query_content):
            start = time.monotonic()
            content = obj.get('content')
            self.process_queue = list()
            self.responses = list()
            if content:
                try:
                    for name in pip_req.parse_requirements(content):
                        logger.info("searching pkg:`{}` in Python Package Index \
                                Repository" .format(name))
                        self.async_fetch(base_url_pypi.format(pkg=name), others=name)
                except Exception as _exc:
                    logger.error("IGNORE: {}".format(_exc))
                    logger.error("Failed to parse content data {}".format(content))

                try:
                    while not self.is_fetch_done(lambda x: x.result().status_code):
                        # hold the process until all request finishes.
                        time.sleep(0.001)
                except Exception as _exc:
                    logger.error("IGNORE: {}".format(_exc))
                    # discard process_queue
                    self.process_queue = []
                    self.responses = []
                packages = sorted(set(self.handle_response()))
                if packages:
                    pkg_string = ', '.join(packages)
                    logger.info("PACKAGES: {}".format(pkg_string))
                    self.counter.update([pkg_string])
                logger.info("Processed content in time: {} process:{}/{}".format(
                    (time.monotonic() - start), idx, len(self.big_query_content)))
        logger.info("Processed All the manifests in time: {}".format(
            time.monotonic() - start_process_time))

        logger.info("updating file content")
        self.update_s3_bucket(data={'pypi': dict(self.counter.most_common())},
                              bucket_name=self.bucket_name,
                              filename=self.filename)

        logger.info("Succefully Processed the PyPiBigQuery")
Process Pypi Bigquery response data.
Process Pypi Bigquery response data.
Process Pypi Bigquery response data .
def process(self):
        start = time.monotonic()
        logger.info("Running Bigquery for pypi synchronously")
        self.big_query_instance.run_query_sync()
        logger.info("fetching bigquery result.")
        for content in self.big_query_instance.get_result():
            self.big_query_content.append(content)
            logger.info("collected manifests: {}".format(len(self.big_query_content)))
        logger.info("Succefully retrieved data from Bigquery, time:{}".format(
            time.monotonic() - start))
        base_url_pypi = 'https://pypi.org/pypi/{pkg}/json'
        logger.info("Starting package cleaning")
        start_process_time = time.monotonic()
        for idx, obj in enumerate(self.big_query_content):
            start = time.monotonic()
            content = obj.get('content')
            self.process_queue = list()
            self.responses = list()
            if content:
                try:
                    for name in pip_req.parse_requirements(content):
                        logger.info("searching pkg:`{}` in Python Package Index \
                                Repository" .format(name))
                        self.async_fetch(base_url_pypi.format(pkg=name), others=name)
                except Exception as _exc:
                    logger.error("IGNORE: {}".format(_exc))
                    logger.error("Failed to parse content data {}".format(content))
                try:
                    while not self.is_fetch_done(lambda x: x.result().status_code):
                        time.sleep(0.001)
                except Exception as _exc:
                    logger.error("IGNORE: {}".format(_exc))
                    self.process_queue = []
                    self.responses = []
                packages = sorted(set(self.handle_response()))
                if packages:
                    pkg_string = ', '.join(packages)
                    logger.info("PACKAGES: {}".format(pkg_string))
                    self.counter.update([pkg_string])
                logger.info("Processed content in time: {} process:{}/{}".format(
                    (time.monotonic() - start), idx, len(self.big_query_content)))
        logger.info("Processed All the manifests in time: {}".format(
            time.monotonic() - start_process_time))
        logger.info("updating file content")
        self.update_s3_bucket(data={'pypi': dict(self.counter.most_common())},
                              bucket_name=self.bucket_name,
                              filename=self.filename)
        logger.info("Succefully Processed the PyPiBigQuery")
def process ( self ) : start = time . monotonic ( ) logger . info ( "Running Bigquery for pypi synchronously" ) self . big_query_instance . run_query_sync ( ) logger . info ( "fetching bigquery result." ) for content in self . big_query_instance . get_result ( ) : self . big_query_content . append ( content ) logger . info ( "collected manifests: {}" . format ( len ( self . big_query_content ) ) ) logger . info ( "Succefully retrieved data from Bigquery, time:{}" . format ( time . monotonic ( ) - start ) ) base_url_pypi = 'https://pypi.org/pypi/{pkg}/json' logger . info ( "Starting package cleaning" ) start_process_time = time . monotonic ( ) for idx , obj in enumerate ( self . big_query_content ) : start = time . monotonic ( ) content = obj . get ( 'content' ) self . process_queue = list ( ) self . responses = list ( ) if content : try : for name in pip_req . parse_requirements ( content ) : logger . info ( "searching pkg:`{}` in Python Package Index \
                                Repository" . format ( name ) ) self . async_fetch ( base_url_pypi . format ( pkg = name ) , others = name ) except Exception as _exc : logger . error ( "IGNORE: {}" . format ( _exc ) ) logger . error ( "Failed to parse content data {}" . format ( content ) ) try : while not self . is_fetch_done ( lambda x : x . result ( ) . status_code ) : time . sleep ( 0.001 ) except Exception as _exc : logger . error ( "IGNORE: {}" . format ( _exc ) ) self . process_queue = [ ] self . responses = [ ] packages = sorted ( set ( self . handle_response ( ) ) ) if packages : pkg_string = ', ' . join ( packages ) logger . info ( "PACKAGES: {}" . format ( pkg_string ) ) self . counter . update ( [ pkg_string ] ) logger . info ( "Processed content in time: {} process:{}/{}" . format ( ( time . monotonic ( ) - start ) , idx , len ( self . big_query_content ) ) ) logger . info ( "Processed All the manifests in time: {}" . format ( time . monotonic ( ) - start_process_time ) ) logger . info ( "updating file content" ) self . update_s3_bucket ( data = { 'pypi' : dict ( self . counter . most_common ( ) ) } , bucket_name = self . bucket_name , filename = self . filename ) logger . info ( "Succefully Processed the PyPiBigQuery" )
Process Pypi Bigquery response data.
Process Pypi Bigquery response data .
"""Process Pypi Bigquery response data.""" # hold the process until all request finishes. # discard process_queue
self