libs/file_processing/file_processing_core.py
def upload_binified_data(binified_data, error_handler, survey_id_dict):
    """ Takes in binified csv data and handles uploading/downloading+updating
        older data to/from S3 for each chunk.
        Returns a set of concatenations that have succeeded and can be removed.
        Returns the number of failed FTPS so that we don't retry them.
        Returns the earliest and latest time bins handled
        Raises any errors on the passed in ErrorHandler."""
    failed_ftps = set([])
    ftps_to_retire = set([])
    upload_these = []

    # Track the earliest and latest time bins, to return them at the end of the function
    earliest_time_bin = None
    latest_time_bin = None

    for data_bin, (data_rows_list, ftp_list) in binified_data.items():
        with error_handler:
            try:
                study_object_id, user_id, data_type, time_bin, original_header = data_bin
                # Update earliest and latest time bins
                if earliest_time_bin is None or time_bin < earliest_time_bin:
                    earliest_time_bin = time_bin
                if latest_time_bin is None or time_bin > latest_time_bin:
                    latest_time_bin = time_bin
                # data_rows_list may be a generator; here it is evaluated
                rows = data_rows_list
                updated_header = convert_unix_to_human_readable_timestamps(original_header, rows)
                chunk_path = construct_s3_chunk_path(study_object_id, user_id, data_type, time_bin)

                if ChunkRegistry.objects.filter(chunk_path=chunk_path).exists():
                    chunk = ChunkRegistry.objects.get(chunk_path=chunk_path)
                    try:
                        s3_file_data = s3_retrieve(chunk_path, study_object_id, raw_path=True)
                    except ReadTimeoutError as e:
                        # The following check was correct for boto 2, still need to hit with boto3 test.
                        if "The specified key does not exist." == str(e):
                            # This error can only occur if the processing gets actually interrupted and
                            # data files fail to upload after DB entries are created.
                            # Encountered this condition 11pm feb 7 2016, cause unknown, there was
                            # no python stacktrace.  Best guess is mongo blew up.
                            # If this happened, delete the ChunkRegistry and push this file upload to the next cycle
                            chunk.remove()  # this line of code is ancient and almost definitely wrong.
                            raise ChunkFailedToExist("chunk %s does not actually point to a file, deleting DB entry, should run correctly on next index." % chunk_path)
                        raise  # Raise original error if not 404 s3 error

                    old_header, old_rows = csv_to_list(s3_file_data)

                    if old_header != updated_header:
                        # To handle the case where a file was on an hour boundary and placed in
                        # two separate chunks we need to raise an error in order to retire this file. If this
                        # happens AND ONE of the files DOES NOT have a header mismatch this may (
                        # will?) cause data duplication in the chunked file whenever the file
                        # processing occurs run.
                        raise HeaderMismatchException('%s\nvs.\n%s\nin\n%s' %
                                                      (old_header, updated_header, chunk_path) )
                    old_rows = list(old_rows)
                    old_rows.extend(rows)
                    del rows
                    ensure_sorted_by_timestamp(old_rows)
                    new_contents = construct_csv_string(updated_header, old_rows)
                    del old_rows

                    upload_these.append((chunk, chunk_path, compress(new_contents), study_object_id))
                    del new_contents
                else:
                    ensure_sorted_by_timestamp(rows)
                    new_contents = construct_csv_string(updated_header, rows)
                    if data_type in SURVEY_DATA_FILES:
                        # We need to keep a mapping of files to survey ids, that is handled here.
                        survey_id_hash = study_object_id, user_id, data_type, original_header
                        survey_id = Survey.objects.filter(
                            object_id=survey_id_dict[survey_id_hash]).values_list("pk", flat=True).get()
                    else:
                        survey_id = None

                    # this object will eventually get **kwarg'd into ChunkRegistry.register_chunked_data
                    chunk_params = {
                        "study_id": Study.objects.filter(object_id=study_object_id).values_list("pk", flat=True).get(),
                        "participant_id": Participant.objects.filter(patient_id=user_id).values_list("pk", flat=True).get(),
                        "data_type": data_type,
                        "chunk_path": chunk_path,
                        "time_bin": time_bin,
                        "survey_id": survey_id
                    }

                    upload_these.append((chunk_params, chunk_path, compress(new_contents), study_object_id))
            except Exception as e:
                # Here we catch any exceptions that may have arisen, as well as the ones that we raised
                # ourselves (e.g. HeaderMismatchException). Whichever FTP we were processing when the
                # exception was raised gets added to the set of failed FTPs.
                failed_ftps.update(ftp_list)
                print(e)
                print("FAILED TO UPDATE: study_id:%s, user_id:%s, data_type:%s, time_bin:%s, header:%s "
                      % (study_object_id, user_id, data_type, time_bin, updated_header))
                raise

            else:
                # If no exception was raised, the FTP has completed processing. Add it to the set of
                # retireable (i.e. completed) FTPs.
                ftps_to_retire.update(ftp_list)

    pool = ThreadPool(CONCURRENT_NETWORK_OPS)
    errors = pool.map(batch_upload, upload_these, chunksize=1)
    for err_ret in errors:
        if err_ret['exception']:
            print(err_ret['traceback'])
            raise err_ret['exception']

    pool.close()
    pool.terminate()
    # The things in ftps to retire that are not in failed ftps.
    # len(failed_ftps) will become the number of files to skip in the next iteration.
    return ftps_to_retire.difference(failed_ftps), len(failed_ftps), earliest_time_bin, latest_time_bin
 Takes in binified csv data and handles uploading/downloading+updating
        older data to/from S3 for each chunk.
        Returns a set of concatenations that have succeeded and can be removed.
        Returns the number of failed FTPS so that we don't retry them.
        Returns the earliest and latest time bins handled
        Raises any errors on the passed in ErrorHandler.
Takes in binified csv data and handles uploading/downloading+updating
older data to/from S3 for each chunk.
Returns a set of concatenations that have succeeded and can be removed.
Returns the number of failed FTPS so that we don't retry them.
Returns the earliest and latest time bins handled
Raises any errors on the passed in ErrorHandler.
Takes in binified csv data and handles uploading / downloading + updating older data to / from S3 for each chunk . Returns a set of concatenations that have succeeded and can be removed . Returns the number of failed FTPS so that we don ' t retry them . Returns the earliest and latest time bins handled Raises any errors on the passed in ErrorHandler .
def upload_binified_data(binified_data, error_handler, survey_id_dict):
    failed_ftps = set([])
    ftps_to_retire = set([])
    upload_these = []
    earliest_time_bin = None
    latest_time_bin = None
    for data_bin, (data_rows_list, ftp_list) in binified_data.items():
        with error_handler:
            try:
                study_object_id, user_id, data_type, time_bin, original_header = data_bin
                if earliest_time_bin is None or time_bin < earliest_time_bin:
                    earliest_time_bin = time_bin
                if latest_time_bin is None or time_bin > latest_time_bin:
                    latest_time_bin = time_bin
                rows = data_rows_list
                updated_header = convert_unix_to_human_readable_timestamps(original_header, rows)
                chunk_path = construct_s3_chunk_path(study_object_id, user_id, data_type, time_bin)
                if ChunkRegistry.objects.filter(chunk_path=chunk_path).exists():
                    chunk = ChunkRegistry.objects.get(chunk_path=chunk_path)
                    try:
                        s3_file_data = s3_retrieve(chunk_path, study_object_id, raw_path=True)
                    except ReadTimeoutError as e:
                        if "The specified key does not exist." == str(e):
                            chunk.remove()  
                            raise ChunkFailedToExist("chunk %s does not actually point to a file, deleting DB entry, should run correctly on next index." % chunk_path)
                        raise  
                    old_header, old_rows = csv_to_list(s3_file_data)
                    if old_header != updated_header:
                        raise HeaderMismatchException('%s\nvs.\n%s\nin\n%s' %
                                                      (old_header, updated_header, chunk_path) )
                    old_rows = list(old_rows)
                    old_rows.extend(rows)
                    del rows
                    ensure_sorted_by_timestamp(old_rows)
                    new_contents = construct_csv_string(updated_header, old_rows)
                    del old_rows
                    upload_these.append((chunk, chunk_path, compress(new_contents), study_object_id))
                    del new_contents
                else:
                    ensure_sorted_by_timestamp(rows)
                    new_contents = construct_csv_string(updated_header, rows)
                    if data_type in SURVEY_DATA_FILES:
                        survey_id_hash = study_object_id, user_id, data_type, original_header
                        survey_id = Survey.objects.filter(
                            object_id=survey_id_dict[survey_id_hash]).values_list("pk", flat=True).get()
                    else:
                        survey_id = None
                    chunk_params = {
                        "study_id": Study.objects.filter(object_id=study_object_id).values_list("pk", flat=True).get(),
                        "participant_id": Participant.objects.filter(patient_id=user_id).values_list("pk", flat=True).get(),
                        "data_type": data_type,
                        "chunk_path": chunk_path,
                        "time_bin": time_bin,
                        "survey_id": survey_id
                    }
                    upload_these.append((chunk_params, chunk_path, compress(new_contents), study_object_id))
            except Exception as e:
                failed_ftps.update(ftp_list)
                print(e)
                print("FAILED TO UPDATE: study_id:%s, user_id:%s, data_type:%s, time_bin:%s, header:%s "
                      % (study_object_id, user_id, data_type, time_bin, updated_header))
                raise
            else:
                ftps_to_retire.update(ftp_list)
    pool = ThreadPool(CONCURRENT_NETWORK_OPS)
    errors = pool.map(batch_upload, upload_these, chunksize=1)
    for err_ret in errors:
        if err_ret['exception']:
            print(err_ret['traceback'])
            raise err_ret['exception']
    pool.close()
    pool.terminate()
    return ftps_to_retire.difference(failed_ftps), len(failed_ftps), earliest_time_bin, latest_time_bin
def upload_binified_data ( binified_data , error_handler , survey_id_dict ) : failed_ftps = set ( [ ] ) ftps_to_retire = set ( [ ] ) upload_these = [ ] earliest_time_bin = None latest_time_bin = None for data_bin , ( data_rows_list , ftp_list ) in binified_data . items ( ) : with error_handler : try : study_object_id , user_id , data_type , time_bin , original_header = data_bin if earliest_time_bin is None or time_bin < earliest_time_bin : earliest_time_bin = time_bin if latest_time_bin is None or time_bin > latest_time_bin : latest_time_bin = time_bin rows = data_rows_list updated_header = convert_unix_to_human_readable_timestamps ( original_header , rows ) chunk_path = construct_s3_chunk_path ( study_object_id , user_id , data_type , time_bin ) if ChunkRegistry . objects . filter ( chunk_path = chunk_path ) . exists ( ) : chunk = ChunkRegistry . objects . get ( chunk_path = chunk_path ) try : s3_file_data = s3_retrieve ( chunk_path , study_object_id , raw_path = True ) except ReadTimeoutError as e : if "The specified key does not exist." == str ( e ) : chunk . remove ( ) raise ChunkFailedToExist ( "chunk %s does not actually point to a file, deleting DB entry, should run correctly on next index." % chunk_path ) raise old_header , old_rows = csv_to_list ( s3_file_data ) if old_header != updated_header : raise HeaderMismatchException ( '%s\nvs.\n%s\nin\n%s' % ( old_header , updated_header , chunk_path ) ) old_rows = list ( old_rows ) old_rows . extend ( rows ) del rows ensure_sorted_by_timestamp ( old_rows ) new_contents = construct_csv_string ( updated_header , old_rows ) del old_rows upload_these . append ( ( chunk , chunk_path , compress ( new_contents ) , study_object_id ) ) del new_contents else : ensure_sorted_by_timestamp ( rows ) new_contents = construct_csv_string ( updated_header , rows ) if data_type in SURVEY_DATA_FILES : survey_id_hash = study_object_id , user_id , data_type , original_header survey_id = Survey . objects . filter ( object_id = survey_id_dict [ survey_id_hash ] ) . values_list ( "pk" , flat = True ) . get ( ) else : survey_id = None chunk_params = { "study_id" : Study . objects . filter ( object_id = study_object_id ) . values_list ( "pk" , flat = True ) . get ( ) , "participant_id" : Participant . objects . filter ( patient_id = user_id ) . values_list ( "pk" , flat = True ) . get ( ) , "data_type" : data_type , "chunk_path" : chunk_path , "time_bin" : time_bin , "survey_id" : survey_id } upload_these . append ( ( chunk_params , chunk_path , compress ( new_contents ) , study_object_id ) ) except Exception as e : failed_ftps . update ( ftp_list ) print ( e ) print ( "FAILED TO UPDATE: study_id:%s, user_id:%s, data_type:%s, time_bin:%s, header:%s " % ( study_object_id , user_id , data_type , time_bin , updated_header ) ) raise else : ftps_to_retire . update ( ftp_list ) pool = ThreadPool ( CONCURRENT_NETWORK_OPS ) errors = pool . map ( batch_upload , upload_these , chunksize = 1 ) for err_ret in errors : if err_ret [ 'exception' ] : print ( err_ret [ 'traceback' ] ) raise err_ret [ 'exception' ] pool . close ( ) pool . terminate ( ) return ftps_to_retire . difference ( failed_ftps ) , len ( failed_ftps ) , earliest_time_bin , latest_time_bin
Takes in binified csv data and handles uploading/downloading+updating
older data to/from S3 for each chunk.
Takes in binified csv data and handles uploading / downloading + updating older data to / from S3 for each chunk .
""" Takes in binified csv data and handles uploading/downloading+updating
        older data to/from S3 for each chunk.
        Returns a set of concatenations that have succeeded and can be removed.
        Returns the number of failed FTPS so that we don't retry them.
        Returns the earliest and latest time bins handled
        Raises any errors on the passed in ErrorHandler.""" # Track the earliest and latest time bins, to return them at the end of the function # Update earliest and latest time bins # data_rows_list may be a generator; here it is evaluated # The following check was correct for boto 2, still need to hit with boto3 test. # This error can only occur if the processing gets actually interrupted and # data files fail to upload after DB entries are created. # Encountered this condition 11pm feb 7 2016, cause unknown, there was # no python stacktrace.  Best guess is mongo blew up. # If this happened, delete the ChunkRegistry and push this file upload to the next cycle # this line of code is ancient and almost definitely wrong. # Raise original error if not 404 s3 error # To handle the case where a file was on an hour boundary and placed in # two separate chunks we need to raise an error in order to retire this file. If this # happens AND ONE of the files DOES NOT have a header mismatch this may ( # will?) cause data duplication in the chunked file whenever the file # processing occurs run. # We need to keep a mapping of files to survey ids, that is handled here. # this object will eventually get **kwarg'd into ChunkRegistry.register_chunked_data # Here we catch any exceptions that may have arisen, as well as the ones that we raised # ourselves (e.g. HeaderMismatchException). Whichever FTP we were processing when the # exception was raised gets added to the set of failed FTPs. # If no exception was raised, the FTP has completed processing. Add it to the set of # retireable (i.e. completed) FTPs. # The things in ftps to retire that are not in failed ftps. # len(failed_ftps) will become the number of files to skip in the next iteration.
binified_data
error_handler
survey_id_dict