Python Multithreading Techniques for Concurrent API Calls, File Downloads, Test Execution, and Database Inserts

Multithreading allows a program to run multiple threads simultaneously, each executing independently; in Python this is achieved via the threading module, which offers functions for creating and managing threads.

Thread Creation and Start – A Thread object is instantiated with a target function and started using start() . Synchronization and mutual exclusion are handled with locks, conditions, or semaphores to avoid race conditions.

Thread Communication and Data Sharing – Threads can share global variables or thread‑safe structures such as Queue , but proper locking is required to ensure safety. Thread lifecycle methods like is_alive() and join() help monitor and control execution.

Example: Concurrent API Requests

import threading
import requests

def send_request(url):
    # send API request
    response = requests.get(url)
    print(f"Response from {url}: {response.text}
")

def main():
    urls = [
        "http://api.example.com/endpoint1",
        "http://api.example.com/endpoint2",
        "http://api.example.com/endpoint3",
    ]
    threads = []
    for url in urls:
        t = threading.Thread(target=send_request, args=(url,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All requests completed.")

if __name__ == "__main__":
    main()

Example: Concurrent File Downloads

import threading
import requests

def download_file(url, filename):
    response = requests.get(url)
    with open(filename, "wb") as file:
        file.write(response.content)
    print(f"Downloaded {filename}")

def main():
    urls = [
        ("http://example.com/file1.pdf", "file1.pdf"),
        ("http://example.com/file2.pdf", "file2.pdf"),
        ("http://example.com/file3.pdf", "file3.pdf"),
    ]
    threads = []
    for url, filename in urls:
        t = threading.Thread(target=download_file, args=(url, filename))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All files downloaded.")

if __name__ == "__main__":
    main()

Example: Concurrent Test Case Execution

import threading
from mytestframework import TestRunner

def run_test_case(test_case):
    result = TestRunner.run(test_case)
    print(f"Test case {test_case} result: {result}")

def main():
    test_cases = ["test_case1", "test_case2", "test_case3"]
    threads = []
    for test_case in test_cases:
        t = threading.Thread(target=run_test_case, args=(test_case,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All test cases executed.")

if __name__ == "__main__":
    main()

Example: Concurrent File Reading

import threading

def read_file(filename):
    with open(filename, "r") as file:
        content = file.read()
    print(f"Content of {filename}: {content}")

def main():
    filenames = ["file1.txt", "file2.txt", "file3.txt"]
    threads = []
    for filename in filenames:
        t = threading.Thread(target=read_file, args=(filename,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All files read.")

if __name__ == "__main__":
    main()

Example: Concurrent Database Inserts

import threading
import sqlite3

def insert_data(data):
    connection = sqlite3.connect("database.db")
    cursor = connection.cursor()
    cursor.execute("INSERT INTO table_name (data) VALUES (?)", (data,))
    connection.commit()
    connection.close()
    print(f"Data {data} inserted into database.")

def main():
    data_list = ["data1", "data2", "data3"]
    threads = []
    for data in data_list:
        t = threading.Thread(target=insert_data, args=(data,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All data inserted into database.")

if __name__ == "__main__":
    main()

The examples demonstrate how Python multithreading can be applied to various automation tasks, improving efficiency by executing I/O‑bound operations concurrently.