libs/file_processing/file_processing_core.py
def process_one_file(
        file_for_processing: FileForProcessing, survey_id_dict: dict, all_binified_data: DefaultDict,
        ftps_to_remove: set
):
    """ This function is the inner loop of the chunking process. """

    if file_for_processing.exception:
        file_for_processing.raise_data_processing_error()

    # there are two cases: chunkable data that can be stuck into "time bins" for each hour, and
    # files that do not need to be "binified" and pretty much just go into the ChunkRegistry unmodified.
    if file_for_processing.chunkable:
        newly_binified_data, survey_id_hash = process_csv_data(file_for_processing)

        # survey answers store the survey id in the file name (truly ancient design decision).
        if file_for_processing.data_type in SURVEY_DATA_FILES:
            survey_id_dict[survey_id_hash] = resolve_survey_id_from_file_name(
                file_for_processing.file_to_process.s3_file_path)

        if newly_binified_data:
            append_binified_csvs(
                all_binified_data, newly_binified_data, file_for_processing.file_to_process
            )
        else:  # delete empty files from FilesToProcess
            ftps_to_remove.add(file_for_processing.file_to_process.id)
        return

    else:
        # case: unchunkable data file
        timestamp = clean_java_timecode(
            file_for_processing.file_to_process.s3_file_path.rsplit("/", 1)[-1][:-4]
        )
        # Since we aren't binning the data by hour, just create a ChunkRegistry that
        # points to the already existing S3 file.
        try:
            ChunkRegistry.register_unchunked_data(
                file_for_processing.data_type,
                timestamp,
                file_for_processing.file_to_process.s3_file_path,
                file_for_processing.file_to_process.study.pk,
                file_for_processing.file_to_process.participant.pk,
                file_for_processing.file_contents,
            )
            ftps_to_remove.add(file_for_processing.file_to_process.id)
        except ValidationError as ve:
            if len(ve.messages) != 1:
                # case: the error case (below) is very specific, we only want that singular error.
                raise

            # case: an unchunkable file was re-uploaded, causing a duplicate file path collision
            # we detect this specific case and update the registry with the new file size
            # (hopefully it doesn't actually change)
            if 'Chunk registry with this Chunk path already exists.' in ve.messages:
                ChunkRegistry.update_registered_unchunked_data(
                    file_for_processing.data_type,
                    file_for_processing.file_to_process.s3_file_path,
                    file_for_processing.file_contents,
                )
                ftps_to_remove.add(file_for_processing.file_to_process.id)
            else:
                # any other errors, add
                raise
 This function is the inner loop of the chunking process. 
This function is the inner loop of the chunking process.
This function is the inner loop of the chunking process .
def process_one_file(
        file_for_processing: FileForProcessing, survey_id_dict: dict, all_binified_data: DefaultDict,
        ftps_to_remove: set
):
    if file_for_processing.exception:
        file_for_processing.raise_data_processing_error()
    if file_for_processing.chunkable:
        newly_binified_data, survey_id_hash = process_csv_data(file_for_processing)
        if file_for_processing.data_type in SURVEY_DATA_FILES:
            survey_id_dict[survey_id_hash] = resolve_survey_id_from_file_name(
                file_for_processing.file_to_process.s3_file_path)
        if newly_binified_data:
            append_binified_csvs(
                all_binified_data, newly_binified_data, file_for_processing.file_to_process
            )
        else:  
            ftps_to_remove.add(file_for_processing.file_to_process.id)
        return
    else:
        timestamp = clean_java_timecode(
            file_for_processing.file_to_process.s3_file_path.rsplit("/", 1)[-1][:-4]
        )
        try:
            ChunkRegistry.register_unchunked_data(
                file_for_processing.data_type,
                timestamp,
                file_for_processing.file_to_process.s3_file_path,
                file_for_processing.file_to_process.study.pk,
                file_for_processing.file_to_process.participant.pk,
                file_for_processing.file_contents,
            )
            ftps_to_remove.add(file_for_processing.file_to_process.id)
        except ValidationError as ve:
            if len(ve.messages) != 1:
                raise
            if 'Chunk registry with this Chunk path already exists.' in ve.messages:
                ChunkRegistry.update_registered_unchunked_data(
                    file_for_processing.data_type,
                    file_for_processing.file_to_process.s3_file_path,
                    file_for_processing.file_contents,
                )
                ftps_to_remove.add(file_for_processing.file_to_process.id)
            else:
                raise
def process_one_file ( file_for_processing : FileForProcessing , survey_id_dict : dict , all_binified_data : DefaultDict , ftps_to_remove : set ) : if file_for_processing . exception : file_for_processing . raise_data_processing_error ( ) if file_for_processing . chunkable : newly_binified_data , survey_id_hash = process_csv_data ( file_for_processing ) if file_for_processing . data_type in SURVEY_DATA_FILES : survey_id_dict [ survey_id_hash ] = resolve_survey_id_from_file_name ( file_for_processing . file_to_process . s3_file_path ) if newly_binified_data : append_binified_csvs ( all_binified_data , newly_binified_data , file_for_processing . file_to_process ) else : ftps_to_remove . add ( file_for_processing . file_to_process . id ) return else : timestamp = clean_java_timecode ( file_for_processing . file_to_process . s3_file_path . rsplit ( "/" , 1 ) [ - 1 ] [ : - 4 ] ) try : ChunkRegistry . register_unchunked_data ( file_for_processing . data_type , timestamp , file_for_processing . file_to_process . s3_file_path , file_for_processing . file_to_process . study . pk , file_for_processing . file_to_process . participant . pk , file_for_processing . file_contents , ) ftps_to_remove . add ( file_for_processing . file_to_process . id ) except ValidationError as ve : if len ( ve . messages ) != 1 : raise if 'Chunk registry with this Chunk path already exists.' in ve . messages : ChunkRegistry . update_registered_unchunked_data ( file_for_processing . data_type , file_for_processing . file_to_process . s3_file_path , file_for_processing . file_contents , ) ftps_to_remove . add ( file_for_processing . file_to_process . id ) else : raise
This function is the inner loop of the chunking process.
This function is the inner loop of the chunking process .
""" This function is the inner loop of the chunking process. """ # there are two cases: chunkable data that can be stuck into "time bins" for each hour, and # files that do not need to be "binified" and pretty much just go into the ChunkRegistry unmodified. # survey answers store the survey id in the file name (truly ancient design decision). # delete empty files from FilesToProcess # case: unchunkable data file # Since we aren't binning the data by hour, just create a ChunkRegistry that # points to the already existing S3 file. # case: the error case (below) is very specific, we only want that singular error. # case: an unchunkable file was re-uploaded, causing a duplicate file path collision # we detect this specific case and update the registry with the new file size # (hopefully it doesn't actually change) # any other errors, add
file_for_processing
FileForProcessing
survey_id_dict
dict
all_binified_data
DefaultDict
ftps_to_remove
set