stress_check.py

import threading
import pandas as pd
import tqdm
import time
import uuid

from app import init
import utils


def main():
    records = init()
    times_data = {'User Creation Time': [], 'Dataset Upload Time': [], 'Total Time': []}
    threads = []
    num_threads = 10
    sample_per_thread = 3
    with tqdm.tqdm(total=num_threads * sample_per_thread) as pbar:
        for _ in range(num_threads):
            thread = threading.Thread(target=thread_action, args=(records, times_data, sample_per_thread, pbar))
            threads.append(thread)
            thread.start()
        for thread in threads:
            thread.join()
    display_statistics(times_data)


def thread_action(records, times_data, sample_per_thread, pbar):
    for _ in range(sample_per_thread):
        seed = str(uuid.uuid4())
        user_creation_time, user = timed_execution(utils.create_new_user, seed, seed)
        dataset_upload_time, _ = timed_execution(utils.upload_dataset, user, seed, seed, records)
        update_times(times_data, user_creation_time, dataset_upload_time)
        pbar.update()


def timed_execution(func, *args, **kwargs):
    start_time = time.time()
    result = func(*args, **kwargs)
    end_time = time.time()
    return end_time - start_time, result


def update_times(times_data, user_creation_time, dataset_upload_time):
    times_data['User Creation Time'].append(user_creation_time)
    times_data['Dataset Upload Time'].append(dataset_upload_time)
    times_data['Total Time'].append(user_creation_time + dataset_upload_time)


def display_statistics(times_data):
    for time_type, times in times_data.items():
        df = pd.DataFrame(times, columns=[time_type])
        print(f"{time_type} Statistics:")
        print(df.describe())


if __name__ == "__main__":
    main()