How to Validate Data with ThreadPoolExecutor in Python
This code shows how to validate a list of numbers concurrently using ThreadPoolExecutor, dramatically speeding up slow validation tasks by running them in parallel threads.
Python code
31 linesimport time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
@dataclass
class Result:
is_valid: bool
value: int
def validate(value: int) -> Result:
time.sleep(0.1) # simulate slow validation (API call, DB check)
return Result(is_valid=0 < value < 100, value=value)
def validate_all(numbers: list) -> list[Result]:
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(validate, numbers)
return list(results)
if __name__ == "__main__":
numbers = [50, 150, 0, 99, -10, 25]
start = time.perf_counter()
results = validate_all(numbers)
elapsed = time.perf_counter() - start
for r in results:
print(f"{r.value} -> valid: {r.is_valid}")
print(f"Validated {len(numbers)} items in {elapsed:.2f}s")Output
50 -> valid: True
150 -> valid: False
0 -> valid: False
99 -> valid: True
-10 -> valid: False
25 -> valid: True
Validated 6 items in 0.10sHow it works
The ThreadPoolExecutor manages a pool of worker threads, so multiple validate() calls run concurrently rather than sequentially. The executor.map() method applies the validation function to each number in the input list and returns results in the same order. Each call to validate() sleeps for 0.1 seconds, so with 4 workers, 6 items complete in roughly 2 batches, taking about 0.2 seconds instead of 0.6 seconds sequentially. The with block ensures the executor properly shuts down and cleans up threads after all tasks complete. The Result dataclass bundles the validation boolean and the original value together for clear, type-safe output.
Common mistakes
- Forgetting that `executor.map` returns results lazily, so converting to a list is required to actually run all tasks.
- Using `ThreadPoolExecutor` for CPU-bound work — threads share the GIL and won't speed up pure computation.
- Assuming results come back in sorted order — `map` preserves input order, but `as_completed` does not.
- Not using a `with` block, leaking executor threads when exceptions occur.
Variations
- Use `executor.submit()` with `as_completed()` if you need to process results as they finish rather than in input order.
- Use `ProcessPoolExecutor` instead when the validation work is CPU-bound and doesn't involve I/O waits.
Real-world use cases
- Validation webhooks or API responses where each payload requires a slow external lookup before being accepted.
- Batch-checking user inputs or form fields against a third-party service like an email or phone verification API.
- Data pipeline preprocessing where each record needs a slow remote check (e.g., geolocation or fraud scoring) before further stages.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.