services/celery_data_processing.py
def celery_process_file_chunks(participant_id):
    """ This is the function is queued up, it runs through all new uploads from a specific user and
    'chunks' them. Handles logic for skipping bad files, raising errors. """

    # celery doesn't clean up after itself very well, either memory or open network connections.
    # this probably has something to do with the fact that celery forks, so possibly picking
    # a different mode would impact this.  Or we can just exit the python process.
    try:
        time_start = datetime.now()
        participant = Participant.objects.get(id=participant_id)

        number_bad_files = 0
        tags = {'user_id': participant.patient_id}
        error_sentry = make_error_sentry(sentry_type=SentryTypes.data_processing)
        print("processing files for %s" % participant.patient_id)

        while True:
            previous_number_bad_files = number_bad_files
            starting_length = participant.files_to_process.exclude(deleted=True).count()

            print("%s processing %s, %s files remaining" % (datetime.now(), participant.patient_id, starting_length))
            number_bad_files += do_process_user_file_chunks(
                    page_size=FILE_PROCESS_PAGE_SIZE,
                    error_handler=error_sentry,
                    position=number_bad_files,
                    participant=participant,
            )
            # If no files were processed, quit processing
            if participant.files_to_process.exclude(deleted=True).count() == starting_length:
                if previous_number_bad_files == number_bad_files:
                    # 2 Cases:
                    #   1) every file broke, blow up. (would cause infinite loop otherwise).
                    #   2) no new files.
                    break
                else:
                    continue

            # put maximum time limit per user
            if (time_start - datetime.now()).total_seconds() > 60*60*3:
                break
    except Exception as e:
        print(f"Error running data processing: {e}")
    finally:
        print(
            "Data processing task completed. Exiting to clean up memory. You can safely ignore the "
            "immediately following \"Worker exited prematurely: exitcode 0\" error message."
        )
        exit(0)
 This is the function is queued up, it runs through all new uploads from a specific user and
    'chunks' them. Handles logic for skipping bad files, raising errors. 
This is the function is queued up, it runs through all new uploads from a specific user and
'chunks' them. Handles logic for skipping bad files, raising errors.
This is the function is queued up it runs through all new uploads from a specific user and ' chunks ' them . Handles logic for skipping bad files raising errors .
def celery_process_file_chunks(participant_id):
    try:
        time_start = datetime.now()
        participant = Participant.objects.get(id=participant_id)
        number_bad_files = 0
        tags = {'user_id': participant.patient_id}
        error_sentry = make_error_sentry(sentry_type=SentryTypes.data_processing)
        print("processing files for %s" % participant.patient_id)
        while True:
            previous_number_bad_files = number_bad_files
            starting_length = participant.files_to_process.exclude(deleted=True).count()
            print("%s processing %s, %s files remaining" % (datetime.now(), participant.patient_id, starting_length))
            number_bad_files += do_process_user_file_chunks(
                    page_size=FILE_PROCESS_PAGE_SIZE,
                    error_handler=error_sentry,
                    position=number_bad_files,
                    participant=participant,
            )
            if participant.files_to_process.exclude(deleted=True).count() == starting_length:
                if previous_number_bad_files == number_bad_files:
                    break
                else:
                    continue
            if (time_start - datetime.now()).total_seconds() > 60*60*3:
                break
    except Exception as e:
        print(f"Error running data processing: {e}")
    finally:
        print(
            "Data processing task completed. Exiting to clean up memory. You can safely ignore the "
            "immediately following \"Worker exited prematurely: exitcode 0\" error message."
        )
        exit(0)
def celery_process_file_chunks ( participant_id ) : try : time_start = datetime . now ( ) participant = Participant . objects . get ( id = participant_id ) number_bad_files = 0 tags = { 'user_id' : participant . patient_id } error_sentry = make_error_sentry ( sentry_type = SentryTypes . data_processing ) print ( "processing files for %s" % participant . patient_id ) while True : previous_number_bad_files = number_bad_files starting_length = participant . files_to_process . exclude ( deleted = True ) . count ( ) print ( "%s processing %s, %s files remaining" % ( datetime . now ( ) , participant . patient_id , starting_length ) ) number_bad_files += do_process_user_file_chunks ( page_size = FILE_PROCESS_PAGE_SIZE , error_handler = error_sentry , position = number_bad_files , participant = participant , ) if participant . files_to_process . exclude ( deleted = True ) . count ( ) == starting_length : if previous_number_bad_files == number_bad_files : break else : continue if ( time_start - datetime . now ( ) ) . total_seconds ( ) > 60 * 60 * 3 : break except Exception as e : print ( f"Error running data processing: {e}" ) finally : print ( "Data processing task completed. Exiting to clean up memory. You can safely ignore the " "immediately following \"Worker exited prematurely: exitcode 0\" error message." ) exit ( 0 )
This is the function is queued up, it runs through all new uploads from a specific user and
'chunks' them.
This is the function is queued up it runs through all new uploads from a specific user and ' chunks ' them .
""" This is the function is queued up, it runs through all new uploads from a specific user and
    'chunks' them. Handles logic for skipping bad files, raising errors. """ # celery doesn't clean up after itself very well, either memory or open network connections. # this probably has something to do with the fact that celery forks, so possibly picking # a different mode would impact this.  Or we can just exit the python process. # If no files were processed, quit processing # 2 Cases: #   1) every file broke, blow up. (would cause infinite loop otherwise). #   2) no new files. # put maximum time limit per user
participant_id