How to try a function with a timeout exception in python

Question

I have a function I want to call in helpers.py that if it runs for more than 10seconds will timeout. I have tried multiprocessing to do this, but I have no idea how to pull my results dictionary out of multiprocessing and was wondering if you all have a good way to do this?

helpers.py

def speed_test():
    print('running speedtest')
    servers = []
    # If you want to test against a specific server
    # servers = [1234]

    threads = None
    # If you want to use a single threaded test
    # threads = 1

    s = speedtest.Speedtest()
    s.get_servers(servers)
    s.get_best_server()
    s.download(threads=threads)
    s.upload(threads=threads)

    results_dict = s.results.dict()
    print('speedtest complete')
    return results_dict

route in flask

from app import helpers

@bp.route('/speed_test')
@login_required
def speed_test1():
    if not current_user.admin:
        flash('Page unacessible.')
        return redirect(url_for('main.index'))

    #This gives my results in a dictionary i can use just fine
    #results = helpers.speed_test()

    #This executes my function just fine, but I have no idea how to pull the dicitonary result from my function
    results = multiprocessing.Process(target=helpers.speed_test)
    results.start()
    results.join(30)
    if results.is_alive():
        print('timeout kill')
        results.kill()
        results='timeout'
    results.join()
    #results.('i want my dictionary from the function return')
    return render_template('test/speedtest.html', title='Speed Test', results=results)

Answer 1

Although a processing pool is normally used for running multiple jobs in parallel, it might be the simplest way of "killing two birds with one stone", that is having a mechanism for timing out the process and for returning a dictionary back from the process:

def speed_test(): # unchanged
    # code omitted
    results_dict = s.results.dict()
    print('speedtest complete')
    return results_dict


@bp.route('/speed_test')
@login_required
def speed_test1():
    #code omitted
    with multiprocessing.Pool(1) as pool: # create a pool size of 1
        result = pool.apply_async(speed_test)
        # the following will generate a multiprocessing.TimeoutError if speed_test takes longer than 10 seconds:
        results_dict = result.get(10)