Python Multiprocessing for Parallel Tasks and API Automation

Multiprocessing refers to running multiple independent processes simultaneously, each with its own memory space and system resources, and in Python it can be implemented using the multiprocessing module.

Compared with multithreading, multiprocessing offers independent memory spaces, system resource isolation, and better CPU utilization, making it suitable for parallel computation on multi‑core CPUs.

In interface automation, multiprocessing can be applied to large‑scale concurrent API testing, accelerate data processing and analysis, enable distributed testing frameworks, run independent tasks in parallel, and execute multiple test suites simultaneously.

Example 1: Concurrent Task Execution

The following code creates four processes that each run process_task with a unique task ID.

import multiprocessing

def process_task(task_id):
    print(f"Task {task_id} is being executed.")

if __name__ == "__main__":
    num_processes = 4
    processes = []
    for i in range(num_processes):
        p = multiprocessing.Process(target=process_task, args=(i,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print("All tasks completed.")

Example 2: Parallel Square Calculation

This snippet launches a separate process for each number to compute its square.

import multiprocessing

def calculate_square(number):
    square = number ** 2
    print(f"Square of {number} is {square}")

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    processes = []
    for number in numbers:
        p = multiprocessing.Process(target=calculate_square, args=(number,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print("All calculations completed.")

Example 3: Parallel Data Processing

Each process converts a string to uppercase, demonstrating simple data transformation in parallel.

import multiprocessing

def process_data(data):
    processed_data = data.upper()
    print(f"Processed data: {processed_data}")

if __name__ == "__main__":
    data_list = ["apple", "banana", "orange", "grape"]
    processes = []
    for data in data_list:
        p = multiprocessing.Process(target=process_data, args=(data,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print("Data processing completed.")

Example 4: Concurrent API Requests

Multiple processes send HTTP GET requests to different URLs in parallel.

import multiprocessing
import requests

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

if __name__ == "__main__":
    urls = ["http://api.example.com/endpoint1", "http://api.example.com/endpoint2"]
    processes = []
    for url in urls:
        p = multiprocessing.Process(target=send_request, args=(url,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print("All requests completed.")

Example 5: Parallel Test Case Execution

This example runs several test cases concurrently, each in its own process.

import multiprocessing

def run_test_case(test_case):
    # Execute test case logic here
    print(f"Running test case: {test_case}")

if __name__ == "__main__":
    test_cases = ["Test Case 1", "Test Case 2", "Test Case 3"]
    processes = []
    for test_case in test_cases:
        p = multiprocessing.Process(target=run_test_case, args=(test_case,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print("All test cases executed.")

These five examples illustrate different scenarios where multiprocessing can improve performance and efficiency in real‑world API automation tasks; adapt and optimize the code according to specific requirements.