O problema da programação síncrona

Introdução à programação assíncrona

Conceitos-chave do asyncio

1. async e await

• async: define uma função como corrotina, permitindo que ela pause sua execução.

• await: suspende a execução da corrotina atual até que outra tarefa seja concluída.

import asyncio

async def main():
    print("Início")
    await asyncio.sleep(2)
    print("Fim")

asyncio.run(main())

Início
(Faz uma pausa de 2 segundos)
Fim

2. Event Loop (loop de eventos)

3. asyncio.run()

Exemplos práticos com asyncio

Simulação de múltiplas requisições web

import asyncio
import time

async def fetch_data(url):
    print(f"Fetching {url}")
    await asyncio.sleep(2)  # Simula uma requisição
    print(f"Finished {url}")

async def main():
    urls = ["http://example.com/1", "http://example.com/2", "http://example.com/3"]
    tasks = [fetch_data(url) for url in urls]
    await asyncio.gather(*tasks)

start_time = time.time()
asyncio.run(main())
print(f"Tempo total: {time.time() - start_time:.2f} segundos")

Fetching http://example.com/1
Fetching http://example.com/2
Fetching http://example.com/3
Finished http://example.com/1
Finished http://example.com/2
Finished http://example.com/3
Tempo total: 2.00 segundos

Uso com bibliotecas de terceiros

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(f"Status: {response.status}")
            return await response.text()

async def main():
    urls = ["https://jsonplaceholder.typicode.com/posts/1", "https://jsonplaceholder.typicode.com/posts/2"]
    tasks = [fetch(url) for url in urls]
    await asyncio.gather(*tasks)

asyncio.run(main())

Esse exemplo destaca o principal benefício da programação assíncrona: a habilidade de gerenciar múltiplas tarefas de I/O ao mesmo tempo, como requisições HTTP, sem bloquear a execução do programa. Assim, você obtém resultados mais rápidos e aproveita melhor os recursos do sistema.

Tratamento de exceções em código assíncrono

async def fetch_data():
    try:
        await asyncio.sleep(1)
        raise ValueError("Erro simulado")
    except ValueError as e:
        print(f"Ocorreu um erro: {e}")

asyncio.run(fetch_data())

Essa abordagem é essencial em programação assíncrona, pois garante que erros em uma corrotina sejam gerenciados corretamente, evitando que eles interrompam todo o fluxo do programa. Assim, você mantém a estabilidade e previsibilidade da aplicação, mesmo diante de falhas inesperadas.

Dicas e boas práticas para asyncio no Python

1. Consistência assíncrona

import asyncio
import requests  # Biblioteca síncrona

async def buscar_dados(url):
    resposta = requests.get(url)  # Operação síncrona bloqueante
    return resposta.text

async def main():
    dados = await buscar_dados("https://www.exemplo.com")
    print(dados[:100])

asyncio.run(main())

import asyncio
import aiohttp  # Biblioteca assíncrona

async def buscar_dados(session, url):
    async with session.get(url) as resposta:
        return await resposta.text()

async def main():
    async with aiohttp.ClientSession() as session:
        dados = await buscar_dados(session, "https://www.exemplo.com")
        print(dados[:100])

asyncio.run(main())

Nesse caso, usamos aiohttp, uma biblioteca assíncrona que mantém o loop de eventos funcionando eficientemente.

2. Utilize bibliotecas assíncronas

• aiohttp: para requisições HTTP.

• asyncpg ou aiomysql: para interações com bancos de dados PostgreSQL ou MySQL.

• asyncios.open_connection(): para trabalhar com sockets de forma assíncrona.

3. Tratamento de exceções

import asyncio
import aiohttp

async def buscar_dados(session, url):
    try:
        async with session.get(url) as resposta:
            resposta.raise_for_status()  # Levanta erro HTTP, se houver
            return await resposta.text()
    except aiohttp.ClientError as e:
        print(f"Erro ao acessar {url}: {e}")
        return None

async def main():
    async with aiohttp.ClientSession() as session:
        dados = await buscar_dados(session, "https://www.naoexiste.com")
        print(dados)

asyncio.run(main())

Neste exemplo, usamos raise_for_status() para capturar erros HTTP e tratamos a exceção de forma que a aplicação não seja interrompida.

4. Gerenciamento de tempo limite (timeouts)

import asyncio
import aiohttp

async def buscar_dados(session, url):
    try:
        async with session.get(url, timeout=5) as resposta:
            return await resposta.text()
    except asyncio.TimeoutError:
        print(f"Timeout ao acessar {url}")
        return None

async def main():
    async with aiohttp.ClientSession() as session:
        dados = await buscar_dados(session, "https://www.exemplo.com")
        print(dados)

asyncio.run(main())

Aqui, o parâmetro timeout=5 limita o tempo de espera a 5 segundos, melhorando a confiabilidade do programa.

5. Testes e monitoramento

• Logs detalhados: registre o fluxo de execução para identificar problemas.

• Ferramentas de profiling: analise onde o tempo está sendo gasto nas operações.

• Uso do asyncio.gather(): execute múltiplas corrotinas concorrentemente e receba os resultados em uma lista.

• Depuradores: familiarize-se com ferramentas como pdb ou depuradores IDE que suportem asyncio.

import asyncio
import logging

logging.basicConfig(level=logging.INFO)

async def tarefa(nome, tempo):
    logging.info(f"{nome} começou")
    await asyncio.sleep(tempo)
    logging.info(f"{nome} terminou")

async def main():
    await asyncio.gather(
        tarefa("Tarefa 1", 2),
        tarefa("Tarefa 2", 1)
    )

asyncio.run(main())

INFO:root:Tarefa 1 começou
INFO:root:Tarefa 2 começou
INFO:root:Tarefa 2 terminou
INFO:root:Tarefa 1 terminou

Use o asyncio.gather com cuidado

async def tarefa(nome, falhar=False):
    if falhar:
        raise ValueError(f"Erro na {nome}")
    return f"{nome} concluída"

async def main():
    tarefas = [
        tarefa("Tarefa 1"),
        tarefa("Tarefa 2", falhar=True),
        tarefa("Tarefa 3")
    ]
    resultados = await asyncio.gather(*tarefas, return_exceptions=True)
    print(resultados)

asyncio.run(main())

['Tarefa 1 concluída', ValueError('Erro na Tarefa 2'), 'Tarefa 3 concluída']

Seguindo essas dicas e práticas, você estará bem preparado para desenvolver aplicações assíncronas robustas, eficientes e escaláveis com asyncio no Python!

Conclusão

Minicurso gratuito de Python: programação web com Flask | Rocketseat

Aprenda a desenvolver um projeto web completo com Python e Flask, de forma prática e gratuita.

https://app.rocketseat.com.br/journey/introducao-ao-python/overview?utm_source=platform&utm_medium=organic&utm_campaign=lead&utm_term=artigo&utm_content=link-asyncio_no_python

Não espere mais! Comece sua jornada com o curso gratuito e mergulhe de cabeça na Formação Python para se tornar o dev que o mercado procura. Vamos juntos?