aiohttp no Python: Guia de Requisições HTTP Assíncronas

•

Cansado de programar?

Conheça a melhor e mais completa formação de Python e Django e sinta-se um programador verdadeiramente competente. Além de Python e Django, você também vai aprender Banco de Dados, SQL, HTML, CSS, Javascript, Bootstrap e muito mais!

Quero aprender Python e Django de Verdade! Quero aprender!
Suporte

Tire suas dúvidas diretamente com o professor

Projetos práticos

Projetos práticos voltados para o mercado de trabalho

Prática profissional

Formação moderna com foco na prática profissional

Resposta rápida

aiohttp é a biblioteca de HTTP assíncrono do Python, feita sobre o asyncio. Instale com pip install aiohttp, abra uma aiohttp.ClientSession() com async with e faça await resp.json(). Com asyncio.gather, dezenas de requisições esperam ao mesmo tempo em vez de uma depois da outra.



1
2
3
4
5
6
7
8
9
10
11
12
13
14



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        # servidor local de testes (código completo mais abaixo)
        async with session.get("http://localhost:9000/produtos/1") as resp:
            print(resp.status)        # 200
            print(await resp.json())  # {'id': 1, 'nome': 'Produto 1', 'preco': 10}


asyncio.run(main())

Resumo em 30 segundos:

  • Crie uma ClientSession e reutilize: ela guarda o pool de conexões.
  • resp.json(), resp.text() e resp.read() são coroutines: precisam de await.
  • 404 e 500 não geram exceção: use resp.raise_for_status().
  • asyncio.gather ou asyncio.TaskGroup (3.11+) disparam várias requisições juntas.
  • Limite a concorrência com asyncio.Semaphore ou TCPConnector(limit=...).

Salve salve Pythonista!

Imagine consultar 20 produtos em uma API que demora meio segundo por resposta. Com requests, o programa faz uma chamada, espera, faz a próxima… e leva 10 segundos. Com o aiohttp, as 20 chamadas esperam ao mesmo tempo e tudo termina em meio segundo.

Você vai usar o aiohttp como cliente HTTP (sessão, JSON, timeout, concorrência, erros e download) e, no final, ver o aiohttp como servidor, erros comuns e exercícios resolvidos.

Os exemplos foram executados com aiohttp 3.14.3 (a versão mais recente no PyPI em setembro de 2026) e Python 3.12, contra um servidor local, para que as suas saídas sejam iguais às do post. Se você ainda não conhece async/await, leia antes o post sobre programação assíncrona com asyncio.

Então… Bora pro post! :rocket:

Vá Direto ao Assunto…

O que é aiohttp e como instalar

O que é aiohttp? aiohttp é uma biblioteca externa de Python para HTTP assíncrono, construída sobre o asyncio. Ela tem duas partes: um cliente (aiohttp.ClientSession), para fazer requisições sem bloquear o programa, e um servidor (aiohttp.web), para criar APIs e sites. O ponto forte é executar muitas requisições de forma concorrente em uma única thread.

A diferença para o requests está na espera. Uma requisição HTTP passa quase todo o tempo aguardando a rede. No código síncrono, o programa fica parado junto. No assíncrono, cada await devolve o controle ao event loop, que aproveita para enviar outras requisições. É concorrência, e não paralelismo: tudo roda em uma única thread, o que funciona porque o gargalo é a rede, e não a CPU. O post sobre concorrência e paralelismo em Python explica essa diferença com calma.

O aiohttp não vem com o Python. Crie um ambiente virtual (ou use o uv) e instale:



1



pip install aiohttp

Para conferir a versão instalada:



1
2
3



import aiohttp

print(aiohttp.__version__)





1



3.14.3

A série 3.14 exige Python 3.10 ou mais novo. A documentação oficial fica em docs.aiohttp.org.

Servidor local para acompanhar os exemplos

APIs públicas mudam e ficam lentas, e aí as saídas deixam de bater. Por isso, os exemplos usam um servidor local escrito com o próprio aiohttp, com atrasos propositais que simulam uma API lenta. Salve como servidor.py:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53



import asyncio

from aiohttp import web

rotas = web.RouteTableDef()


@rotas.get("/produtos/{id}")
async def produto(request):
    produto_id = int(request.match_info["id"])
    if produto_id > 100:
        return web.json_response({"erro": "produto não encontrado"}, status=404)
    await asyncio.sleep(0.5)  # simula uma API lenta
    return web.json_response({"id": produto_id, "nome": f"Produto {produto_id}", "preco": produto_id * 10})


@rotas.get("/atraso/{segundos}")
async def atraso(request):
    segundos = float(request.match_info["segundos"])
    await asyncio.sleep(segundos)
    return web.json_response({"atraso": segundos})


@rotas.get("/busca")
async def busca(request):
    return web.json_response({
        "params": dict(request.query),
        "user_agent": request.headers.get("User-Agent"),
    })


@rotas.post("/pedidos")
async def criar_pedido(request):
    dados = await request.json()
    return web.json_response({"id": 1, **dados}, status=201)


@rotas.get("/status/{codigo}")
async def status(request):
    codigo = int(request.match_info["codigo"])
    return web.Response(status=codigo, text=f"status {codigo}")


@rotas.get("/arquivo")
async def arquivo(request):
    return web.Response(body=b"x" * 1_000_000, content_type="application/octet-stream")


app = web.Application()
app.add_routes(rotas)

if __name__ == "__main__":
    web.run_app(app, host="localhost", port=9000)

Rode em um terminal e deixe aberto:



1



python servidor.py





1
2



======== Running on http://localhost:9000 ========
(Press CTRL+C to quit)

O essencial: /produtos/{id} demora 0,5 segundo (404 na hora para ids acima de 100) e /atraso/{segundos} demora o tempo pedido. Se a porta 9000 estiver ocupada, troque. O código é explicado na seção sobre aiohttp.web.

ClientSession: a peça central do aiohttp

No aiohttp, toda requisição passa por uma ClientSession, que guarda o pool de conexões, os cabeçalhos padrão, os cookies e o timeout.



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/produtos/7") as resp:
            print(resp.status, resp.reason, resp.ok)
            print(resp.headers["Content-Type"])
            print(resp.url)
            dados = await resp.json()
            print(dados["nome"], dados["preco"])


asyncio.run(main())





1
2
3
4



200 OK True
application/json; charset=utf-8
http://localhost:9000/produtos/7
Produto 7 70

Repare nos dois async with:

  • async with aiohttp.ClientSession() as session: abre a sessão e garante que as conexões sejam fechadas no final. Sem isso, o aiohttp avisa Unclosed client session.
  • async with session.get(...) as resp: envia a requisição e devolve a conexão ao pool no fim do bloco. Leia o corpo (await resp.json()) dentro dele.

O asyncio.run(main()) inicia o event loop (o padrão antigo com get_event_loop() quebra no Python 3.14).

Por que reutilizar a mesma sessão

A documentação oficial é direta: não crie uma sessão por requisição. Abrir uma conexão custa caro (TCP e, em HTTPS, o handshake TLS). A sessão mantém conexões abertas (keep-alive) e as reaproveita. O recomendado é uma sessão por aplicação, passada como argumento para as funções que fazem requisições, como nos exemplos a seguir.

Cuidado: a sessão só vive dentro do async with. Se uma função a cria com async with e a devolve com return, a próxima requisição lança RuntimeError: Session is closed.

GET, POST, JSON, params e headers com aiohttp

Leitura da resposta

A regra do resp (um ClientResponse): o que depende de ler o corpo precisa de await.

Atributo ou método O que devolve Precisa de await?
resp.status O código HTTP (200, 404…) Não
resp.ok True se o código for menor que 400 Não
resp.headers Os cabeçalhos da resposta Não
resp.url A URL final, com a query string Não
await resp.json() O corpo JSON como dict ou list Sim
await resp.text() O corpo como str Sim
await resp.read() O corpo como bytes Sim

params, headers e base_url

Filtros na query string vão em params. Cabeçalhos que valem para todas as chamadas vão em headers na sessão. E base_url evita repetir o endereço do servidor:



1
2
3
4
5
6
7
8
9
10
11
12
13
14



import asyncio

import aiohttp


async def main():
    cabecalhos = {"User-Agent": "meu-app/1.0"}
    async with aiohttp.ClientSession(base_url="http://localhost:9000", headers=cabecalhos) as session:
        async with session.get("/busca", params={"q": "teclado mecânico", "pagina": 2}) as resp:
            print(resp.url)
            print(await resp.json())


asyncio.run(main())





1
2



http://localhost:9000/busca?q=teclado+mec%C3%A2nico&pagina=2
{'params': {'q': 'teclado mecânico', 'pagina': '2'}, 'user_agent': 'meu-app/1.0'}

O aiohttp codificou o espaço e o acento, e o servidor recebeu tudo como texto ('2'). Se o base_url tiver um caminho, ele precisa terminar com / (como "https://api.exemplo.com/v1/"), senão a sessão lança ValueError: base_url must have a trailing '/'; nesse caso, use caminhos sem a barra inicial (session.get("produtos/7")). Para um token, passe headers={"Authorization": f"Bearer {token}"} na sessão ou na requisição.

POST com json=

Para enviar JSON no corpo, use json=. O aiohttp serializa o dicionário e define o cabeçalho Content-Type: application/json:



1
2
3
4
5
6
7
8
9
10
11
12
13
14



import asyncio

import aiohttp


async def main():
    novo_pedido = {"produto_id": 7, "quantidade": 2}
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        async with session.post("/pedidos", json=novo_pedido) as resp:
            print(resp.status)
            print(await resp.json())


asyncio.run(main())





1
2



201
{'id': 1, 'produto_id': 7, 'quantidade': 2}

Para formulário, use data=. session.put(), session.patch() e session.delete() seguem o mesmo formato. Para revisar a conversão entre JSON e Python, veja como manipular JSON no Python.

raise_for_status e ClientTimeout

raise_for_status: 404 e 500 não são exceção

Como no requests, 404 ou 500 é uma resposta válida: o aiohttp não lança erro sozinho. Para transformar 4xx e 5xx em exceção, chame resp.raise_for_status():



1
2
3
4
5
6
7
8
9
10
11
12
13
14



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        async with session.get("/produtos/999") as resp:
            print(resp.status, resp.ok)
            resp.raise_for_status()
            print("esta linha não executa")


asyncio.run(main())





1
2
3
4



404 False
Traceback (most recent call last):
  ...
aiohttp.client_exceptions.ClientResponseError: 404, message='Not Found', url='http://localhost:9000/produtos/999'

Para valer em todas as chamadas, passe raise_for_status=True na sessão. A ClientResponseError traz o código em .status e a mensagem em .message:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000", raise_for_status=True) as session:
        try:
            async with session.get("/produtos/999") as resp:
                print(await resp.json())
        except aiohttp.ClientResponseError as erro:
            print(f"Falhou com status {erro.status}: {erro.message}")


asyncio.run(main())





1



Falhou com status 404: Not Found

ClientTimeout: não espere 5 minutos

Diferente do requests, que por padrão espera para sempre, o aiohttp tem timeout padrão: 300 segundos (5 minutos) no total e 30 segundos para abrir o socket. É tempo demais para uma API. Defina o seu com aiohttp.ClientTimeout, na sessão ou em uma requisição:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20



import asyncio

import aiohttp


async def main():
    timeout = aiohttp.ClientTimeout(total=2)
    async with aiohttp.ClientSession(base_url="http://localhost:9000", timeout=timeout) as session:
        try:
            async with session.get("/atraso/5") as resp:
                print(await resp.json())
        except asyncio.TimeoutError as erro:
            print("A API demorou mais de 2 segundos:", type(erro).__name__)

        # timeout específico para uma única requisição
        async with session.get("/atraso/1", timeout=aiohttp.ClientTimeout(total=3)) as resp:
            print(await resp.json())


asyncio.run(main())





1
2



A API demorou mais de 2 segundos: TimeoutError
{'atraso': 1.0}

Além de total, o ClientTimeout aceita connect, sock_connect e sock_read (tempo máximo entre dois pedaços de dados); None ou 0 desliga o limite. Sobre a exceção: no Python 3.11+, asyncio.TimeoutError e o TimeoutError embutido são a mesma classe. No 3.10, o timeout do aiohttp não é capturado por except TimeoutError: (testamos). Por isso os exemplos usam asyncio.TimeoutError, que funciona em todas as versões.

Está curtindo esse conteúdo? :thumbsup:

Que tal receber 30 dias de conteúdo direto na sua Caixa de Entrada?

Sua assinatura não pôde ser validada.
Você fez sua assinatura com sucesso.

Assine as PyDicas e receba 30 dias do melhor conteúdo Python na sua Caixa de Entrada: direto e sem enrolação!

Muitas requisições concorrentes com asyncio.gather e TaskGroup

É aqui que o aiohttp faz diferença. O padrão: uma coroutine que faz uma requisição com a sessão compartilhada, executada várias vezes em conjunto.

asyncio.gather

asyncio.gather recebe várias coroutines, executa todas de forma concorrente e devolve a lista de resultados na mesma ordem em que elas foram passadas:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22



import asyncio
import time

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    inicio = time.perf_counter()
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        produtos = await asyncio.gather(*(buscar_produto(session, i) for i in range(1, 11)))
    duracao = time.perf_counter() - inicio
    print(f"{len(produtos)} produtos em {duracao:.2f}s")
    print([p["nome"] for p in produtos[:3]])


asyncio.run(main())





1
2



10 produtos em 0.51s
['Produto 1', 'Produto 2', 'Produto 3']

Dez requisições de 0,5 segundo terminaram em 0,51 segundo: todas esperaram ao mesmo tempo. O * desempacota o gerador em argumentos, como explicado no post sobre *args e **kwargs.

asyncio.TaskGroup (Python 3.11+)

O asyncio.TaskGroup, do Python 3.11, é a forma mais moderna de rodar tarefas juntas. Você cria as tarefas com tg.create_task() e, ao sair do async with, todas terminaram:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22



import asyncio
import time

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    inicio = time.perf_counter()
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        async with asyncio.TaskGroup() as tg:
            tarefas = [tg.create_task(buscar_produto(session, i)) for i in range(1, 11)]
    produtos = [tarefa.result() for tarefa in tarefas]
    print(f"{len(produtos)} produtos em {time.perf_counter() - inicio:.2f}s")


asyncio.run(main())





1



10 produtos em 0.51s

A diferença aparece quando algo dá errado: se uma tarefa falha, as outras são canceladas e os erros chegam em um ExceptionGroup, tratado com except*:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23



import asyncio

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        try:
            async with asyncio.TaskGroup() as tg:
                tarefas = [tg.create_task(buscar_produto(session, i)) for i in (1, 2, 999)]
        except* aiohttp.ClientResponseError as grupo:
            for erro in grupo.exceptions:
                print("Falhou:", erro.status, erro.request_info.url)
        print([t.cancelled() for t in tarefas])


asyncio.run(main())





1
2



Falhou: 404 http://localhost:9000/produtos/999
[True, True, False]

O produto 999 respondeu 404 na hora, e as buscas 1 e 2, que ainda esperavam, foram canceladas. No gather, a primeira exceção é propagada, mas as outras coroutines continuam rodando. Para coletar sucessos e falhas lado a lado, use return_exceptions=True (veja a seção de erros).

Como limitar a concorrência: Semaphore e TCPConnector

Disparar 5 mil requisições de uma vez contra a API de alguém é pedir para tomar erro 429 (Too Many Requests). Há duas formas de limitar a concorrência.

asyncio.Semaphore

Um asyncio.Semaphore(n) deixa no máximo n coroutines entrarem no bloco async with semaforo: ao mesmo tempo. As outras esperam a vez:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31



import asyncio
import time

import aiohttp

ativas = 0
pico = 0


async def buscar_produto(session, semaforo, produto_id):
    global ativas, pico
    async with semaforo:  # no máximo 5 coroutines passam daqui ao mesmo tempo
        ativas += 1
        pico = max(pico, ativas)
        async with session.get(f"/produtos/{produto_id}") as resp:
            dados = await resp.json()
        ativas -= 1
        return dados


async def main():
    semaforo = asyncio.Semaphore(5)
    inicio = time.perf_counter()
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        produtos = await asyncio.gather(
            *(buscar_produto(session, semaforo, i) for i in range(1, 21))
        )
    print(f"{len(produtos)} produtos em {time.perf_counter() - inicio:.2f}s, pico de {pico} simultâneas")


asyncio.run(main())





1



20 produtos em 2.01s, pico de 5 simultâneas

Vinte requisições em grupos de 5 dão 4 ondas de 0,5 segundo: 2 segundos (ativas e pico só provam o limite). O semáforo é o mais flexível, porque pode envolver qualquer trecho, como a requisição e o processamento da resposta.

TCPConnector(limit=…)

A outra forma é limitar as conexões da própria sessão:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20



import asyncio
import time

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        return await resp.json()


async def main():
    conector = aiohttp.TCPConnector(limit=5)  # no máximo 5 conexões abertas
    inicio = time.perf_counter()
    async with aiohttp.ClientSession(base_url="http://localhost:9000", connector=conector) as session:
        produtos = await asyncio.gather(*(buscar_produto(session, i) for i in range(1, 21)))
    print(f"{len(produtos)} produtos em {time.perf_counter() - inicio:.2f}s")


asyncio.run(main())





1



20 produtos em 2.01s

Os padrões do TCPConnector são limit=100 e limit_per_host=0 (sem limite por servidor). Ou seja, a sessão nunca abre mais de 100 conexões: com 200 requisições ao nosso servidor, o tempo foi de 1,06 segundo, duas ondas de 100. Use limit_per_host quando a sessão fala com vários servidores.

Tratamento de erros no aiohttp

Todas as exceções do cliente herdam de aiohttp.ClientError. As três situações que você mais vai tratar:

Situação Exceção
Servidor respondeu 4xx ou 5xx (com raise_for_status) aiohttp.ClientResponseError
Não conseguiu conectar (DNS, porta fechada, servidor fora do ar) aiohttp.ClientConnectorError
Estourou o ClientTimeout asyncio.TimeoutError
Qualquer falha do cliente aiohttp aiohttp.ClientError

Uma função que trata cada caso sem derrubar as outras requisições:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33



import asyncio

import aiohttp


async def buscar(session, url):
    try:
        async with session.get(url) as resp:
            resp.raise_for_status()
            return await resp.json()
    except aiohttp.ClientResponseError as erro:
        return f"HTTP {erro.status} em {url}"
    except aiohttp.ClientConnectorError:
        return f"Não foi possível conectar em {url}"
    except asyncio.TimeoutError:
        return f"Tempo esgotado em {url}"


async def main():
    urls = [
        "http://localhost:9000/produtos/3",
        "http://localhost:9000/produtos/999",
        "http://localhost:9999/produtos/1",  # porta sem servidor
        "http://localhost:9000/atraso/5",
    ]
    timeout = aiohttp.ClientTimeout(total=2)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        resultados = await asyncio.gather(*(buscar(session, url) for url in urls))
    for resultado in resultados:
        print(resultado)


asyncio.run(main())





1
2
3
4



{'id': 3, 'nome': 'Produto 3', 'preco': 30}
HTTP 404 em http://localhost:9000/produtos/999
Não foi possível conectar em http://localhost:9999/produtos/1
Tempo esgotado em http://localhost:9000/atraso/5

Outra opção é return_exceptions=True no gather: em vez de propagar o primeiro erro, ele coloca a exceção na lista de resultados, na posição da chamada que falhou:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25



import asyncio

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        resultados = await asyncio.gather(
            *(buscar_produto(session, i) for i in (1, 999, 2)),
            return_exceptions=True,
        )
    for resultado in resultados:
        if isinstance(resultado, Exception):
            print("Erro:", type(resultado).__name__, resultado.status)
        else:
            print("OK:", resultado["nome"])


asyncio.run(main())





1
2
3



OK: Produto 1
Erro: ClientResponseError 404
OK: Produto 2

Download de arquivo em chunks (streaming)

await resp.read() carrega o corpo inteiro na memória. Para arquivos grandes, leia em blocos com resp.content.iter_chunked() e grave cada bloco no disco:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23



import asyncio

import aiohttp


async def baixar(session, url, destino):
    total = 0
    async with session.get(url) as resp:
        resp.raise_for_status()
        with open(destino, "wb") as arquivo:
            async for bloco in resp.content.iter_chunked(64 * 1024):  # 64 KB por vez
                arquivo.write(bloco)
                total += len(bloco)
    return total


async def main():
    async with aiohttp.ClientSession() as session:
        total = await baixar(session, "http://localhost:9000/arquivo", "arquivo.bin")
    print(f"{total} bytes salvos em arquivo.bin")


asyncio.run(main())





1



1000000 bytes salvos em arquivo.bin

Cada volta do async for traz no máximo 64 KB, então a memória fica pequena mesmo para arquivos de vários GB. Detalhe: arquivo.write() é síncrono e bloqueia o event loop por um instante a cada bloco. Em um download, tudo bem; com muitos arquivos grandes em paralelo, considere asyncio.to_thread() para a escrita. Para revisar open() e os modos de arquivo, veja como manipular arquivos com Python.

requests vs aiohttp: comparação de tempo real

Vamos medir. O script busca os mesmos 20 produtos (0,5 segundo cada) com requests, em sequência e já com Session, e com aiohttp, de forma concorrente:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34



import asyncio
import time

import aiohttp
import requests

BASE = "http://localhost:9000"
IDS = range(1, 21)  # 20 produtos, cada resposta demora 0,5 s


def com_requests():
    with requests.Session() as sessao:
        return [sessao.get(f"{BASE}/produtos/{i}", timeout=10).json() for i in IDS]


async def buscar(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        return await resp.json()


async def com_aiohttp():
    async with aiohttp.ClientSession(base_url=BASE) as session:
        return await asyncio.gather(*(buscar(session, i) for i in IDS))


inicio = time.perf_counter()
r1 = com_requests()
print(f"requests (sequencial):  {len(r1)} produtos em {time.perf_counter() - inicio:.2f}s")

inicio = time.perf_counter()
r2 = asyncio.run(com_aiohttp())
print(f"aiohttp (concorrente):  {len(r2)} produtos em {time.perf_counter() - inicio:.2f}s")

print("Mesmo resultado?", r1 == r2)





1
2
3



requests (sequencial):  20 produtos em 10.04s
aiohttp (concorrente):  20 produtos em 0.51s
Mesmo resultado? True

Rodamos três vezes e os números se repetiram: cerca de 20 vezes mais rápido, com o mesmo resultado. O tempo do requests é a soma das esperas (20 x 0,5 s); o do aiohttp é praticamente a espera mais longa. Agora, a parte honesta:

  • O ganho vem da espera de rede. Se o servidor responde em 5 ms, a diferença encolhe. Se o gargalo é CPU, async não ajuda: o caminho é multiprocessing.
  • requests com threads também fica concorrente: no mesmo teste, ThreadPoolExecutor(max_workers=20) terminou em 0,52 s. O aiohttp brilha com centenas ou milhares de requisições (uma coroutine custa bem menos que uma thread) e em programas que já são assíncronos.
  • São números de um servidor local com atraso artificial. Contra uma API real, os tempos variam com a rede e com os limites do servidor.

Programação assíncrona só rende de verdade em cima de uma base forte de Python e de APIs bem construídas. Essa fundação, do básico às APIs REST com Django, é o que a Jornada Python constrói com você:

aiohttp, requests ou httpx: qual usar

O httpx é uma alternativa excelente: API quase igual à do requests, modo síncrono (httpx.Client) e assíncrono (httpx.AsyncClient) e suporte a HTTP/2. O aiohttp é mais antigo no ecossistema asyncio, focado em alto volume de requisições assíncronas, e traz um servidor web junto.

Situação Escolha
Script simples, poucas requisições requests
Muitas requisições em código síncrono, sem reescrever para async requests + ThreadPoolExecutor
Centenas ou milhares de requisições concorrentes aiohttp (ou httpx.AsyncClient)
Aplicação já assíncrona (bot, crawler, API async) aiohttp ou httpx.AsyncClient
Mesmo cliente para código síncrono e assíncrono, ou HTTP/2 httpx
Cliente e servidor HTTP assíncronos na mesma biblioteca aiohttp

aiohttp como servidor com aiohttp.web

O aiohttp também é um framework web assíncrono, e o servidor de testes do início já era uma aplicação aiohttp.web. As peças essenciais, segundo o quickstart do servidor:

  • web.RouteTableDef() cria a tabela de rotas; @rotas.get(...) e @rotas.post(...) registram os handlers.
  • Cada handler é uma coroutine que recebe o request e devolve web.Response(text=...) ou web.json_response(dados).
  • request.match_info["id"] lê variáveis do caminho, request.query a query string e await request.json() o corpo JSON.
  • web.run_app(app) sobe o servidor (porta 8080 por padrão).

Uma API de tarefas em memória:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38



from aiohttp import web

rotas = web.RouteTableDef()
tarefas = []


@rotas.get("/")
async def inicio(request):
    return web.Response(text="API de tarefas no ar")


@rotas.get("/tarefas")
async def listar(request):
    return web.json_response(tarefas)


@rotas.post("/tarefas")
async def criar(request):
    dados = await request.json()
    tarefa = {"id": len(tarefas) + 1, "titulo": dados["titulo"]}
    tarefas.append(tarefa)
    return web.json_response(tarefa, status=201)


@rotas.get("/tarefas/{id}")
async def detalhar(request):
    tarefa_id = int(request.match_info["id"])
    for tarefa in tarefas:
        if tarefa["id"] == tarefa_id:
            return web.json_response(tarefa)
    raise web.HTTPNotFound(text="tarefa não encontrada")


app = web.Application()
app.add_routes(rotas)

if __name__ == "__main__":
    web.run_app(app, host="localhost", port=9001)

E um cliente aiohttp conversando com ele:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9001") as session:
        async with session.post("/tarefas", json={"titulo": "Estudar aiohttp"}) as resp:
            print(resp.status, await resp.json())
        async with session.get("/tarefas") as resp:
            print(await resp.json())
        async with session.get("/tarefas/42") as resp:
            print(resp.status, await resp.text())


asyncio.run(main())





1
2
3



201 {'id': 1, 'titulo': 'Estudar aiohttp'}
[{'id': 1, 'titulo': 'Estudar aiohttp'}]
404 tarefa não encontrada

O aiohttp.web é maduro e rápido, ótimo para serviços pequenos, WebSockets e ferramentas internas. Para APIs REST novas com validação e documentação automática, muita gente prefere FastAPI; para sistemas completos com banco, admin e autenticação, o Django segue mais produtivo.

Erros comuns

Os erros mais comuns de quem começa no aiohttp, com a última linha real do traceback.

Esquecer o await em resp.json() (TypeError)



1
2
3
4
5
6
7
8
9
10
11
12
13



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/produtos/1") as resp:
            dados = resp.json()  # faltou o await
            print(dados["nome"])


asyncio.run(main())





1
2



TypeError: 'coroutine' object is not subscriptable
sys:1: RuntimeWarning: coroutine 'ClientResponse.json' was never awaited

Correção: dados = await resp.json(). json(), text() e read() são coroutines: sem await, você recebe o objeto coroutine, não os dados.

Usar with em vez de async with (TypeError)



1
2
3
4
5
6
7
8
9
10
11
12



import asyncio

import aiohttp


async def main():
    with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/produtos/1") as resp:
            print(await resp.json())


asyncio.run(main())





1



TypeError: Use async with instead

Correção: async with aiohttp.ClientSession() as session:. Fechar as conexões é uma operação assíncrona, por isso a sessão exige async with.

Chamar asyncio.run() no Jupyter ou no Colab (RuntimeError)

O Jupyter e o Google Colab já têm um event loop rodando. Chamar asyncio.run() dentro dele equivale a este código:



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/produtos/1") as resp:
            print(await resp.json())


async def simula_jupyter():
    asyncio.run(main())  # o Jupyter já tem um event loop rodando

asyncio.run(simula_jupyter())





1



RuntimeError: asyncio.run() cannot be called from a running event loop

Correção: no notebook, escreva await main() direto na célula. Em arquivos .py, continue com asyncio.run(main()).

Usar o padrão antigo get_event_loop() no Python 3.14 (RuntimeError)



1
2
3
4
5
6
7
8



import asyncio


async def main():
    return 1

loop = asyncio.get_event_loop()
print(loop.run_until_complete(main()))





1



RuntimeError: There is no current event loop in thread 'MainThread'.

Correção: troque as duas linhas por asyncio.run(main()). No Python 3.12 esse código antigo ainda funciona, mas emite DeprecationWarning: There is no current event loop. No 3.14 ele quebra.

Chamar .json() em uma resposta que não é JSON (ContentTypeError)



1
2
3
4
5
6
7
8
9
10
11
12



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/status/200") as resp:
            print(await resp.json())


asyncio.run(main())





1



aiohttp.client_exceptions.ContentTypeError: 200, message='Attempt to decode JSON with unexpected mimetype: text/plain; charset=utf-8', url='http://localhost:9000/status/200'

Correção: o aiohttp confere o Content-Type antes de decodificar. Para texto ou HTML, use await resp.text(). Se a API devolve JSON com o cabeçalho errado, await resp.json(content_type=None) desliga a checagem.

Exercícios resolvidos

Os exercícios usam o servidor local do início, que precisa estar rodando.

Exercício 1. Busque o produto 5 em http://localhost:9000/produtos/5 e mostre a frase Produto 5 custa R$ 50, usando os campos nome e preco da resposta.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:9000/produtos/5") as resp:
            produto = await resp.json()
    print(f"{produto['nome']} custa R$ {produto['preco']}")


asyncio.run(main())

Saída: Produto 5 custa R$ 50

O await resp.json() fica dentro do bloco da requisição; depois, produto é um dicionário comum.

Exercício 2. Envie um POST para /pedidos com o JSON {"produto_id": 3, "quantidade": 4} e mostre o status, o id e a quantidade devolvidos.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13
14



import asyncio

import aiohttp


async def main():
    pedido = {"produto_id": 3, "quantidade": 4}
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        async with session.post("/pedidos", json=pedido) as resp:
            dados = await resp.json()
            print(resp.status, dados["id"], dados["quantidade"])


asyncio.run(main())

Saída: 201 1 4

O json= serializa o dicionário; o status 201 (Created) indica que o recurso foi criado.

Exercício 3. Usando params, faça um GET em /busca com q=python e limite=5. Mostre a URL final e os parâmetros que o servidor recebeu.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13



import asyncio

import aiohttp


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        async with session.get("/busca", params={"q": "python", "limite": 5}) as resp:
            print(resp.url)
            print((await resp.json())["params"])


asyncio.run(main())

Saída:



1
2



http://localhost:9000/busca?q=python&limite=5
{'q': 'python', 'limite': '5'}

Na URL não existe tipo, então o 5 chega ao servidor como a string '5'.

Exercício 4. Busque os produtos de 1 a 5 de forma concorrente com asyncio.gather e mostre a lista de preços e a soma deles.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18



import asyncio

import aiohttp


async def buscar_preco(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return (await resp.json())["preco"]


async def main():
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        precos = await asyncio.gather(*(buscar_preco(session, i) for i in range(1, 6)))
    print(precos, sum(precos))


asyncio.run(main())

Saída: [10, 20, 30, 40, 50] 150

O gather devolve os resultados na ordem das chamadas, não na ordem de chegada das respostas.

Exercício 5. Escreva a coroutine buscar_com_limite(session, url, segundos) que devolve o JSON da resposta ou None se a requisição demorar mais que segundos. Teste com /atraso/3 e /atraso/0.2, ambos com limite de 1 segundo.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20



import asyncio

import aiohttp


async def buscar_com_limite(session, url, segundos):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=segundos)) as resp:
            return await resp.json()
    except asyncio.TimeoutError:
        return None


async def main():
    async with aiohttp.ClientSession() as session:
        print(await buscar_com_limite(session, "http://localhost:9000/atraso/3", 1))
        print(await buscar_com_limite(session, "http://localhost:9000/atraso/0.2", 1))


asyncio.run(main())

Saída:



1
2



None
{'atraso': 0.2}

O ClientTimeout passado na requisição vale só para ela.

Exercício 6. Busque os produtos [1, 150, 3, 404, 5] de forma concorrente, sem deixar que os erros 404 interrompam as outras buscas. Mostre quantos deram certo, quantos falharam e os ids dos que deram certo.

Ver solução 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24



import asyncio

import aiohttp


async def buscar_produto(session, produto_id):
    async with session.get(f"/produtos/{produto_id}") as resp:
        resp.raise_for_status()
        return await resp.json()


async def main():
    ids = [1, 150, 3, 404, 5]
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        resultados = await asyncio.gather(
            *(buscar_produto(session, i) for i in ids), return_exceptions=True
        )
    sucessos = [r for r in resultados if not isinstance(r, Exception)]
    falhas = [r for r in resultados if isinstance(r, Exception)]
    print(f"{len(sucessos)} sucessos, {len(falhas)} falhas")
    print([r["id"] for r in sucessos])


asyncio.run(main())

Saída:



1
2



3 sucessos, 2 falhas
[1, 3, 5]

Com return_exceptions=True, cada ClientResponseError entra na lista no lugar da resposta, e o isinstance separa os dois casos.

Exercício 7 (estilo prova). Sobre a aiohttp.ClientSession, assinale a alternativa correta:

a) O recomendado é criar uma nova sessão para cada requisição, para evitar conflito entre chamadas.
b) A sessão mantém um pool de conexões e deve ser reutilizada em várias requisições.
c) A sessão pode ser aberta com with aiohttp.ClientSession() as session:, como um arquivo.
d) A sessão só permite requisições GET; para POST é preciso usar o requests.

Ver solução

Resposta: b. A sessão reaproveita conexões abertas (keep-alive), por isso a documentação recomenda uma sessão por aplicação. A a é o que a documentação manda evitar, a c lança TypeError: Use async with instead e a d é falsa: a sessão tem post(), put(), delete() e os demais métodos.

Exercício 8 (estilo prova). Cada requisição a /produtos/{id} demora 0,5 segundo. Quanto tempo, aproximadamente, o código abaixo leva para executar?



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21



import asyncio
import time

import aiohttp


async def buscar(session, semaforo, produto_id):
    async with semaforo:
        async with session.get(f"/produtos/{produto_id}") as resp:
            return await resp.json()


async def main():
    semaforo = asyncio.Semaphore(2)
    inicio = time.perf_counter()
    async with aiohttp.ClientSession(base_url="http://localhost:9000") as session:
        await asyncio.gather(*(buscar(session, semaforo, i) for i in range(1, 11)))
    print(f"{time.perf_counter() - inicio:.1f}s")


asyncio.run(main())

a) 0,5 segundo, porque o gather executa tudo ao mesmo tempo
b) 1,0 segundo
c) 2,5 segundos
d) 5,0 segundos, porque o código fica sequencial

Ver solução

Saída: 2.5s

Resposta: c. O semáforo deixa só 2 requisições simultâneas: 10 requisições formam 5 ondas de 0,5 segundo, 2,5 segundos. A a ignora o semáforo e a d seria o tempo sequencial (10 x 0,5).

Quer praticar APIs e automação em projetos guiados? Conheça o nosso curso de Python completo.

Conclusão

Neste guia de aiohttp, você aprendeu:

✅ ClientSession - uma por aplicação, sempre com async with
✅ GET, POST, params, headers e json= - e await resp.json()
✅ raise_for_status e ClientTimeout - erros HTTP viram exceção
✅ gather e TaskGroup - dezenas de requisições no tempo de uma
✅ Semaphore e TCPConnector - concorrência com limite
✅ Erros e streaming - exceções do cliente e download em blocos
✅ aiohttp.web - o mesmo pacote também cria servidores

Próximos passos:

Se ficou com alguma dúvida, fique à vontade para deixar um comentário no box aqui embaixo! Será um prazer te responder! :wink:

Perguntas frequentes

O que é aiohttp no Python?

aiohttp é uma biblioteca externa de Python para HTTP assíncrono, construída sobre o asyncio. Ela funciona como cliente (fazer requisições com aiohttp.ClientSession) e como servidor (criar APIs com aiohttp.web). A grande vantagem é disparar muitas requisições ao mesmo tempo sem threads: enquanto uma resposta não chega, o event loop atende as outras. Instale com pip install aiohttp.

Qual a diferença entre aiohttp e requests?

O requests é síncrono: cada chamada bloqueia o programa até a resposta chegar, então 20 requisições de 0,5 segundo levam cerca de 10 segundos em sequência. O aiohttp é assíncrono: com async/await e asyncio.gather, as mesmas 20 requisições levam cerca de 0,5 segundo, porque todas esperam ao mesmo tempo. Para poucas chamadas em um script simples, o requests é mais fácil. Para dezenas ou milhares de chamadas, ou dentro de uma aplicação que já é assíncrona, use aiohttp.

Por que não criar uma ClientSession para cada requisição no aiohttp?

Porque a ClientSession guarda um pool de conexões. Reutilizar a mesma sessão aproveita conexões já abertas (keep-alive) e evita refazer a conexão TCP e o handshake TLS a cada chamada. A documentação oficial recomenda uma sessão por aplicação, aberta com async with aiohttp.ClientSession() as session: e compartilhada entre todas as requisições.

Como fazer várias requisições ao mesmo tempo com aiohttp?

Crie uma coroutine que faz uma requisição usando a sessão compartilhada e execute várias delas com await asyncio.gather(*(buscar(session, i) for i in ids)), que devolve os resultados na mesma ordem das chamadas. No Python 3.11 ou mais novo você também pode usar async with asyncio.TaskGroup() as tg: e tg.create_task(...), que cancela as tarefas restantes se uma delas falhar.

Como limitar o número de requisições simultâneas no aiohttp?

Há duas formas. Com asyncio.Semaphore(5), envolvendo a requisição em async with semaforo:, no máximo 5 coroutines fazem a chamada ao mesmo tempo. Com aiohttp.TCPConnector(limit=5) passado em ClientSession(connector=...), a sessão abre no máximo 5 conexões. O padrão do TCPConnector é limit=100 no total e limit_per_host=0 (sem limite por host).

Qual o timeout padrão do aiohttp e como mudar?

O timeout padrão da ClientSession é de 300 segundos (5 minutos) no total da operação, com 30 segundos para abrir o socket. Para mudar, passe timeout=aiohttp.ClientTimeout(total=10) na sessão ou em uma requisição específica. Quando o tempo estoura, o aiohttp lança asyncio.TimeoutError, que no Python 3.11 ou mais novo é a mesma classe que o TimeoutError embutido.

aiohttp ou httpx: qual usar?

O httpx oferece uma API parecida com a do requests e funciona tanto de forma síncrona (httpx.get) quanto assíncrona (httpx.AsyncClient), além de suportar HTTP/2. O aiohttp é só assíncrono, é mais antigo e maduro no ecossistema asyncio e também traz um servidor web. Se você quer um único cliente para código síncrono e assíncrono, escolha httpx. Se o projeto é todo assíncrono e precisa de alto volume de requisições ou de um servidor embutido, aiohttp é uma ótima escolha.

Em uma questão de prova, o que acontece ao chamar resp.json() sem await no aiohttp?

No aiohttp, resp.json() é uma coroutine. Sem await, você recebe um objeto coroutine e não o dicionário, e ao tentar acessar uma chave aparece TypeError: 'coroutine' object is not subscriptable, além do aviso RuntimeWarning: coroutine 'ClientResponse.json' was never awaited. O correto é dados = await resp.json(). O mesmo vale para resp.text() e resp.read().

Autor

Por Vinícius Ramos em

Engenheiro de Computação pela UnB com especialização em Inteligência Artificial. Sr. Software Engineer com 10+ anos de experiência desenvolvendo sistemas críticos em Python e Django. Atuei no Banco do Brasil liderando equipes para monitoramento de mainframe, trabalhei como desenvolvedor sênior em projetos forenses para a Polícia Federal, e atualmente sou engenheiro de software na Labcorp (Estados Unidos), desenvolvendo soluções escaláveis para testes genéticos. Fundador da Python Academy, compartilhando conhecimento técnico com a comunidade brasileira desde 2018.

Linkedin Bora se conectar no LinkedIn! vinicius-aramos

Minhas redes:

  • Email
  • Github
  • StackOverflow

Começe agora sua Jornada na Programação!

Não deixe para amanhã o sucesso que você pode começar a construir hoje!

#newsletter Olá :wave: Curtiu o artigo? Então faça parte da nossa Newsletter! Privacidade Não se preocupe, respeitamos sua privacidade. Você pode se descadastrar a qualquer momento.

Continue aprendendo!

🔗 Web Scraping com Python e BeautifulSoup: guia prático com exemplos testados 🔗 O que é WebScraping 🔗 Python e LangChain: Web scraping com IA 🔗 Programação Assíncrona com Python
🔗 Aprenda sobre Concorrência e Paralelismo em Python 🔗 Multithreading no Python: threading, ThreadPoolExecutor e GIL 🔗 Filas em Python (queues) 🔗 PyPy: seu projeto Python mais rápido!