thread_pool_async_jobs.py

import logging
import random

from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from time import sleep, time
from functools import wraps

logging.basicConfig(
    format='%(asctime)s %(levelname)-8s %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S')

def log_completed_with_args(func):
    """wrapper to log function name and args once it is
    completed.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"{func.__name__} called with: \
{args if args else ''} {kwargs if kwargs else ''}")
        result = func(*args, **kwargs)
        logging.info(f"{func.__name__} completed with: \
{args if args else ''} {kwargs if kwargs else ''}")
        return result
    return wrapper

@dataclass
class Job:
    "Mock out a job with an id that takes between 0 and 5 seconds"
    id: int
    def result(self):
        "simulates random runtime between zero and 5 second"
        sleep(random.randint(0, 5))


@log_completed_with_args
def start_long_running_job(i):
    return Job(i)

@log_completed_with_args
def post_process_job(job):
    "Waits for job results."
    job_result = job.result()
    # Do something to job_result

def process_ten_multi_threaded_jobs():
    """Start 10 jobs and process their results using nested
    ThreadPoolExecutor.map()s"""
    with ThreadPoolExecutor() as p:
        p.map(post_process_job, p.map(start_long_running_job, range(10)))

if __name__ == "__main__":
    START = time()
    process_ten_multi_threaded_jobs()
    logging.info(f"Ran ten multithreaded jobs in {time() - START} seconds.")


## Example Logs:
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (0,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (0,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (1,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (1,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (3,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (3,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (2,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (2,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (4,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (4,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (5,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (5,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (6,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (6,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (7,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (7,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (8,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (8,)
# 2019-08-20 21:35:35 INFO     start_long_running_job called with: (9,)
# 2019-08-20 21:35:35 INFO     start_long_running_job completed with: (9,)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=0),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=1),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=2),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=3),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=4),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=5),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=6),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=7),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=8),)
# 2019-08-20 21:35:35 INFO     post_process_job called with: (Job(id=9),)
# 2019-08-20 21:35:36 INFO     post_process_job completed with: (Job(id=4),)
# 2019-08-20 21:35:36 INFO     post_process_job completed with: (Job(id=3),)
# 2019-08-20 21:35:38 INFO     post_process_job completed with: (Job(id=0),)
# 2019-08-20 21:35:38 INFO     post_process_job completed with: (Job(id=5),)
# 2019-08-20 21:35:38 INFO     post_process_job completed with: (Job(id=9),)
# 2019-08-20 21:35:39 INFO     post_process_job completed with: (Job(id=7),)
# 2019-08-20 21:35:40 INFO     post_process_job completed with: (Job(id=1),)
# 2019-08-20 21:35:40 INFO     post_process_job completed with: (Job(id=2),)
# 2019-08-20 21:35:40 INFO     post_process_job completed with: (Job(id=6),)
# 2019-08-20 21:35:40 INFO     post_process_job completed with: (Job(id=8),)
# 2019-08-20 21:35:40 INFO     Ran ten multithreaded jobs in 5.010130882263184 seconds.