Aiohttp.ClientSession.get: примеры (PYTHON)

Примеры использования aiohttp.ClientSession.get

Простой GET-запрос

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/get') as response:
            print(await response.text())

asyncio.run(main())

{
  \"args\": {},
  \"headers\": {
    \"Host\": \"httpbin.org\",
    \"Accept-Encoding\": \"gzip, deflate\"
  },
  \"origin\": \"89.64.12.34\",
  \"url\": \"https://httpbin.org/get\"
}

Запрос с параметрами и заголовками

async with aiohttp.ClientSession() as session:
    params = {'key1': 'value1', 'key2': 'value2'}
    headers = {'User-Agent': 'MyApp/1.0'}
    async with session.get('https://httpbin.org/get',
                           params=params,
                           headers=headers) as response:
        data = await response.json()
        print(data['args'])
        print(data['headers']['User-Agent'])

{'key1': 'value1', 'key2': 'value2'}
MyApp/1.0

Запрос с ограничением по времени

import aiohttp
from aiohttp import ClientTimeout

async with aiohttp.ClientSession() as session:
    timeout = ClientTimeout(total=10)
    async with session.get('https://httpbin.org/delay/5',
                           timeout=timeout) as response:
        print(f'Status: {response.status}')

Status: 200

Альтернативные функции в Python

Синхронная библиотека Requests

Функция requests.get() из одноименной библиотеки предоставляет синхронный интерфейс для HTTP-запросов. Подход подходит для блокирующих операций в синхронном коде.

import requests

response = requests.get('https://httpbin.org/get')
print(response.json())

Асинхронная библиотека HTTPX

Библиотека httpx предлагает как синхронный, так и асинхронный клиент. Метод httpx.AsyncClient.get() является современной альтернативой с поддержкой HTTP/2.

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://httpbin.org/get')
        print(response.json())

asyncio.run(main())

Выбор между aiohttp, httpx и requests зависит от требований проекта. aiohttp оптимален для асинхронных приложений, httpx подходит для современных проектов с HTTP/2, а requests используется в синхронном коде.

Реализации в других языках программирования

JavaScript (Node.js с Axios)

const axios = require('axios');

axios.get('https://httpbin.org/get')
  .then(response => {
    console.log(response.data);
  })
  .catch(error => {
    console.error(error);
  });

Java (HttpClient)

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create(\"https://httpbin.org/get\"))
    .build();

client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
    .thenApply(HttpResponse::body)
    .thenAccept(System.out::println);

Go (net/http)

package main

import (
    \"fmt\"
    \"io/ioutil\"
    \"net/http\"
)

func main() {
    resp, err := http.Get(\"https://httpbin.org/get\")
    if err != nil {
        panic(err)
    }
    defer resp.Body.Close()
    body, _ := ioutil.ReadAll(resp.Body)
    fmt.Println(string(body))
}

Основное отличие Python-реализации состоит в асинхронной модели выполнения, в то время как многие другие языки используют потоки или callback-функции.

Распространенные ошибки

Отсутствие await при чтении ответа

async with session.get('https://httpbin.org/get') as response:
    text = response.text()  # Ошибка: пропущен await
    print(text)

RuntimeWarning: coroutine 'ClientResponse.text' was never awaited

Использование сессии вне контекстного менеджера

session = aiohttp.ClientSession()
response = await session.get('https://httpbin.org/get')
# Забыт вызов session.close()

Такая практика приводит к утечкам соединений. Правильно использовать конструкцию async with.

Игнорирование кодов состояния HTTP

async with session.get('https://httpbin.org/status/404') as response:
    data = await response.json()  # Ошибка для статуса 404

aiohttp.client_exceptions.ContentTypeError

Рекомендуется проверять статус ответа через response.status.

Изменения в последних версиях aiohttp

В версии 3.8.0 представлен параметр raise_for_status для автоматической проверки кодов состояния. При значении True метод выбрасывает исключение для статусов 4xx и 5xx.

async with session.get('https://httpbin.org/status/404',
                       raise_for_status=True) as response:
    data = await response.text()

aiohttp.client_exceptions.ClientResponseError

В версии 3.9.0 улучшена поддержка типизации для методов клиента. Добавлены аннотации типов для аргументов и возвращаемых значений.

Расширенные примеры применения

Параллельные запросы с asyncio.gather

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'https://httpbin.org/get',
        'https://httpbin.org/ip',
        'https://httpbin.org/user-agent'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result[:100])

asyncio.run(main())

Использование прокси-сервера

async with aiohttp.ClientSession() as session:
    proxy = 'http://proxy.example.com:8080'
    async with session.get('https://httpbin.org/get',
                           proxy=proxy) as response:
        print(await response.text())

Кастомные SSL-сертификаты

import ssl

ssl_context = ssl.create_default_context(cafile='path/to/cert.pem')
connector = aiohttp.TCPConnector(ssl=ssl_context)

async with aiohttp.ClientSession(connector=connector) as session:
    async with session.get('https://example.com') as response:
        print(response.status)

Обработка больших ответов по частям

async with session.get('https://httpbin.org/stream/100') as response:
    async for line in response.content:
        if line:
            print(f'Получено: {line.decode().strip()}')
            break

Такой подход эффективен для обработки потоковых данных без загрузки всего ответа в память.