# Multithreading to process requests and save results in python

I was presented with a task to come up with a script that generates a CSV with POSTAL codes via bruteforce (I'm in Brazil, so that means CEP to us).

Points to note:

• I'm using an external library, but basically consuming a SOAP service provided by the Postal agency;
• I have a list of cities I have to get the results for, with variable ranges to be tested;
• I have 8 threads available for me, and using 7 to process and 1 to writing (Iw as scared of sincronism);

When formulating the solution, I had two paths:

1. Send a city to each thread, and wait for that city to finish processing.
2. Send the entire list of postal codes available for a city to all threads, and only grab the next city after it is all done.

Currently I opted for option 1, since I was thinking the number of requests and processing would be the same, but now it does seem a bit slow (it's been running for around 7 days and not done).

Problems:

• There're cities with huge ranges to test (like Sao Paulo, with 6.5M), which means a thread stuck processing it;
• but on the other hand, it's complicated to manage which cities have completed by going for option 2.

My input data is a CSV file, formatted like this (This is in Brazilian Portuguese):

Adamantina,17800-000 a 17809-999,Não codificada por logradouros,Total do município


My code is below. Some of the code can be ignored (like the logging statements):

import threading
import csv
import queue
import time
from datetime import timedelta

import requests
import pycep_correios
from loguru import logger
from pycep_correios import HOMOLOGACAO, PRODUCAO
from tqdm import tqdm

"""Utility function to read CSV files

Arguments:
csv_path {str} -- path to the CSV file

Returns:
list -- a list of lists where each CSV row is a list
"""
with open(csv_path, 'r') as f:
return [list(row) for row in csv_reader]

def recover_range(range_string: str):
"""Utility function to recover the numeric range from string

Given that the string is in a specific format,
we split it and get the numeric range

Arguments:
range_string {str} -- string containing the range, must be in format 'xxxxx-xxx a xxxxx-xxx'

Returns:
list -- list with the numeric range ([from, to])
"""
cep_range = [v.replace('-', '') for v in range_string.split(' ')]
cep_range = [int(v) for v in cep_range if v.isnumeric()]
return cep_range

def __init__(self, queue):
logger.info('Instantiating {} on {}', self.__class__.__name__,
self.name)
self.queue = queue

def write_to_file(self, content: dict):
"""Method to write content to the target CSV

Arguments:
content {dict} -- dictionary where each key is a column for the CSV
"""
with open('output_sp.csv', 'a') as f:
logger.info('{}: writing new register to CSV.', self.name)
writer = csv.DictWriter(f, content.keys())
writer.writerow(content)

def run(self):
while True:
result = self.queue.get()
self.write_to_file(result)

def __init__(self, in_queue: queue.Queue, out_queue: queue.Queue):
logger.info('Instantiating {} on {}', self.__class__.__name__,
self.name)
self.in_queue = in_queue
self.out_queue = out_queue

def run(self):
while True:
record = self.in_queue.get()
for item in record:
result = self.process(item)
if result is not None:
self.out_queue.put(result)
else:

def process(self, record):
"""Processing method

This method expects a CEP number as input, and will make a request
to the Correios API to check if it exists. In positive cases we
return the data, else None is returned.

Arguments:
record {int} -- CEP number

Returns:
dict -- Dictionary containing the valid CEP data, None if not found
"""
cep = str(record)
if len(cep) < 8:
cep = '0{}'.format(cep)
if pycep_correios.validar_cep(cep):
try:
ambiente=HOMOLOGACAO)
logger.success('{}: {} is a valid CEP, saving...',
self.name, cep)
except pycep_correios.excecoes.ExcecaoPyCEPCorreios as exc:
logger.error('{}: Exception when processing: {} - [{}]',
self.name, cep, exc.message)
return None
except AttributeError as exc:
logger.error('{}: CEP {} has AttributeError {}', self.name,
cep, exc)
except requests.exceptions.ConnectionError:
logger.warning(
'{}: Connection blocked, trying again in 60 seconds...',
self.name)
time.sleep(60)  # After some time the SOAP service force disconnects
# Recursive call to reprocess the item
return self.process(cep)
else:
logger.error('{}: Invalid CEP {}', self.name, cep)
return None

def main():
rotation="25 MB",
compression="zip",
format="{time} {level} {message}",
level="INFO")
cep_list = [
item for item in cep_list
]
cep_queue = queue.Queue()  # list of ceps to process
result_queue = queue.Queue()  # results acquired

for _ in range(3):  # 4 cores = 3 proc threads + 1 writer thread
t.daemon = True
t.start()

t.daemon = True
t.start()

cep_processing_list = []
for record in cep_list:
cep_ranges = recover_range(record[1])
if not cep_ranges:  # dealing with header rows
continue
else:
cep_processing_list.append(
[cep for cep in range(cep_ranges[0], cep_ranges[1])])

for record in tqdm(cep_processing_list):
cep_queue.put(record)

start_time = time.time()
logger.info('Starting the script...')

# wait for queue to get empty
cep_queue.join()
result_queue.join()

logger.info('Finished. Done in {}.',
str(timedelta(seconds=(time.time() - start_time))))
logger.remove(lg)

if __name__ == "__main__":
main()



The goals I want to achieve:

• Improve efficiency overall;
• Improve code to be cleaner and more compliant with Python guidelines (I tried my best);
• Get some advice if I should split this code in functions and files more;
• If possible, come up with a solution for the huge amount of data implied.

If more hardware is the only answer, there's nothing I can do. But I would like to do the best I can with the code. Thanks in advance.

EDIT: As this question has been around for some time and has not seem much activity besides the initial response I am marking it as solved, but if there're any pointers you would like to comment, you are free to do.

• While unintuitive, you should know that the threading package does not actually do proper multithreading, it only simulates asynchronous threads. It's great for doing HTTP requests, and waiting for results. But you probably want to look at the multiprocessing package, which actually runs your code in parallel, and offers a speedup. – maxb Sep 3 '19 at 12:50
• That's interesting to know. I am quite new to multithreading in python, thanks for the head up. – inblank Sep 4 '19 at 13:55

## Type hints for return values

You add str as a parameter hint, but you're missing a return value hint - for read_csv for instance, it should be list.

You're also missing a type hint for queue.

Don't pre-convert cep to a string. Instead:

cep = '{:08d}'.format(cep)


## f-strings

logger.success(f'{self.name}: {cep} is a valid CEP, saving...')


## Recursion for retry

Don't! There's no need to recurse, and it's needlessly bloating your stack. Rather than recursing, use a simple loop.

## continue logic

    if not cep_ranges:  # dealing with header rows
continue
else:
cep_processing_list.append(
[cep for cep in range(cep_ranges[0], cep_ranges[1])])


should be

if cep_ranges:
cep_processing_list.append(
[cep for cep in range(cep_ranges[0], cep_ranges[1])])

• Thank you for your pointers, I was not aware loguru supports f-strings directly. The recursion strategy is a hack, and your suggestion seems pretty much smarter. – inblank Aug 30 '19 at 1:26
• An f-string, after formatting, turns into a regular string - so anything that accepts a str supports an f-string. – Reinderien Aug 30 '19 at 1:33
• Overall, what do you think of the logic? – inblank Aug 30 '19 at 12:51