Este post é uma continuidade dos posts Lendo arquivos CSV em Python e fazendo requisições e do post Fazendo requests concorrentemente (assíncronas) no Python
Após fazermos as requests assíncronas no último post obtivemos uma melhoria signifcativa em nosso processamento, conseguindo paralelizar as requests e reduzindo muito o tempo. Porém não estamos fazendo o melhor proveito da API async do Python, percebam que cada função async também é responsável por também imprimir o resultado e o ideal seria retornarmos os resultados para tratar posteriormente. Vamos fazer isso:
async def process_register(register):
async with httpx.AsyncClient() as client:
response = await client.put(
url=f'https://teste-api-async.free.beeceptor.com/clientes/{register["id"]}',
json={
'nome': register['nome']
}
)
status_code = response.status_code
# Adicionei um pequeno tratamento de erro caso o status_code não seja 200
return status_code, response.json() if status_code == 200 else {'result': 'error'}
A mudança foi bem sútil, apenas começamos a retornar o código de status da requisição e o payload. Agora na nossa função main precisamos receber estes resultados e iterar sobre eles.
async def main():
with open('arquivo.csv') as f:
reader = csv.DictReader(f)
tasks = set()
for register in reader:
tasks.add(
asyncio.create_task(process_register(register))
)
done, pending = await asyncio.wait(tasks)
Veja que agora passarmos armazenar o resultado de nossa função em duas variáveis, done e pending. Na variável done estão todas as tasks que foram concluídas e na pending as que ainda não terminaram, como por padrão a função asyncio.wait espera todas concluirem a variável pending sempre retornará um conjunto vazio. Agora o que precisamos fazer é iterar sobre os resultados retornados e fazer o processamento necessário sobre eles:
async def main():
with open('arquivo.csv') as f:
reader = csv.DictReader(f)
tasks = set()
for register in reader:
tasks.add(
asyncio.create_task(process_register(register))
)
done, pending = await asyncio.wait(tasks)
for task in done:
print(task.result()) # Começamos a consumir os resultados aqui!
No código acima iteramos sobre as tasks e passamos vamos coletando todos os resultados através do método .result(). Com isto já temos como processar diversos resultados de uma vez. O único ponto aqui é que como disparamos todas as requisições do nosso CSV de uma vez, pode acontecer de sermos bloqueados pelo serviço (imagine que o CSV tenha vinte mil registros), ou até pior, poderíamos derrubar o serviço em questão como se fosse um ataque DoS. Resolver isso é bem simples, podemos dividir nossa lista em pedaços de cinco (para isso vou utilizar o código deste post do Stackoverflow que deve ser o trecho de código de três linhas mais copiado da história, risos) de tamanho fixo para realizar este processamento e acumular os resultados para o final:
def chunks(lst, n):
"""Retorna um iterador de listas dividindo a lista original em pedaços"""
for i in range(0, len(lst), n):
yield lst[i:i + n]
async def main():
with open('arquivo.csv') as f:
reader = csv.DictReader(f)
registers = list(reader) # Como vamos quebrar em lista menores, vamos converter antes em uma lista
processed = []
for chunk in chunks(registers, 5):
tasks = set()
for register in chunk:
print(register)
tasks.add(
asyncio.create_task(process_register(register))
)
done, pending = await asyncio.wait(tasks)
for task in done:
processed.append(task.result()) # Armazenamos o resultado em uma lista
for p in processed: # Exibimos quando conluir tudo
print(p)
Com isso agora temos um script que faz o processamento em paralelo porém sem oferecer riscos para o serviço que estamos chamando. Podemos aumentar o número de requisições em paralelo de acordo com o serviço e também podemos salvar os resultados em outro arquivo, o que vai depender da sua necessidade, mas o que fizemos até aqui já é uma base que permite criar scripts muito úteis e também aprendemos a utizar as bibliotecas async do Python para fazer requisições em paralelo, podendo utilizar estes conceitos de diversas outras formas.