Improve your APIs Response Time, use threading!

When to use?

In general, threading is a valuable technique to employ in scenarios where your program is not primarily engaged in computational tasks but rather is waiting for some external event, such as

• I/O operations,
• network requests,
• user input,
• waiting for resources to become available

How to apply?

Their are various methods in python for applying threading.
Example:
a. threading library
b. concurrent.futures
c. asyncio and async functions

General Approach

import threading

def download_from_cloud():
    pass

threads = []
for url, filename in zip(urls, filenames):
  # Create and start thread for each download
  thread = threading.Thread(target=download_from_cloud, args=(url, filename))
  threads.append(thread)
  thread.start()

# In the above code the downloading has started concurrently

# Wait for all threads to finish (optional)
for thread in threads:
  thread.join()

• thread.join() will make the program to wait for all the threads to first finish and than continue with the further code. You can also choose to move further without waiting for them to complete depending on your programs dependency on the operations performed by the above thread.

Below is a Helper Module

Past the below code in a .py file

from concurrent.futures import ThreadPoolExecutor
import time

class ConcurrentThreadExecutor:

    def __init__(self):
        pass

    def execute_parallel(self, tasks, max_threads):
        with ThreadPoolExecutor(max_workers=max_threads) as executor:
            results = list(executor.map(lambda task: self.execute_task(*task), tasks))
        return results

    def execute_task(self, task_name, task_function, task_args):
        t1 = time.time()
        print(f"Started ThreadTask : {task_name}")
        result = self.run_function(task_function, *task_args)
        print(f'ThreadTask : "{task_name}" took : {round(time.time() - t1,2)}sec')
        return (task_name, result)

    def run_function(self, func, *args, **kwargs):
        return func(*args, **kwargs)

class MultiThreadExecutor(ConcurrentThreadExecutor):

    def __init__(self, debug: bool = True) -> None:
        self.max_threads = ThreadPoolExecutor()._max_workers 
        if debug:
            print("Initialized Multi-Thread Executor, Number of threads: " + str(self.max_threads))
        self.debug = debug


    def execute_tasks(self, tasks: list, no_of_cores : int = None):
        """
         ** Note Input must Follow the following format **
            tasks : [ ( TASK_NAME, FUNCTION, ARGS ), ]
        """
        num_threads = min(self.max_threads, len(tasks)) if no_of_cores is None else min(no_of_cores, self.max_threads)
        if self.debug:
            print("Number of threads being used: " + str(num_threads))
        results = self.execute_parallel(tasks, num_threads)

        results = dict(results)
        return results

Now just call the MultiThreadExecutor to run multiple functions concurrently as below

def func1(name ):
    time.sleep(5)
    print(f"Completed : {name}")


def func2(name):
    [i*i for i in range(100000)]
    print(f"Completed : {name}")

# Create the MultiThreadExecutor Object
execute_concurrent = MultiThreadExecutor(debug = False)

# Pass All the function to be executed concurrently

execute_concurrent.execute_tasks([
    ('func1', func1, ('Function 1',)),
    ('func2', func2, ('Function 2',))
])

print('Ended!')

Unlocking Efficiency with Threading:

Threading is a powerful tool to enhance responsiveness in Python applications. By allowing your program to handle waiting tasks concurrently, it can deliver a smoother user experience.

Hope you found it informative!