python aiohttp client session get post and json requests: Pr

python aiohttp client session get post and json requests requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

When working with asynchronous HTTP in Python, aiohttp's ClientSession is the core interface for making GET, POST, and JSON requests. Unlike requests, which is synchronous, aiohttp integrates with asyncio and allows you to manage connection pooling, cookies, and timeouts across multiple requests. This article shows how to create a session, send GET and POST requests with JSON payloads, parse JSON responses, and handle common failure modes.

Creating a ClientSession

ClientSession is the recommended way to make HTTP requests with aiohttp. It maintains a connection pool and stores cookies, so you should create one session per application or logical client and reuse it for all requests. The session must be closed when you are done, typically via an async context manager:

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as resp:
            print(resp.status)
            data = await resp.json()
            print(data)

asyncio.run(main())

The outer async with ensures the session is properly closed, releasing the underlying connections. If you create a session without a context manager, you must call await session.close() explicitly. Reusing a session avoids the overhead of establishing a new TCP connection and TLS handshake for every request.

Making GET Requests with Query Parameters

To send a GET request with query parameters, pass a dictionary to the params argument. aiohttp will URL-encode the keys and values automatically:

params = {'q': 'python aiohttp', 'page': 2}
async with session.get('https://api.example.com/search', params=params) as resp:
    data = await resp.json()

The resulting URL becomes https://api.example.com/search?q=python+aiohttp&page=2. You can also pass headers, such as an authorization token, using the headers argument:

headers = {'Authorization': 'Bearer <token>'}
async with session.get(url, headers=headers) as resp:
    ...

Sending POST Requests with JSON Payloads

For POST requests, use the json parameter to send a Python object as the request body. aiohttp serializes it to JSON and sets the Content-Type header to application/json:

payload = {'name': 'Alice', 'age': 30}
async with session.post('https://api.example.com/users', json=payload) as resp:
    result = await resp.json()

If you need to send JSON with a different content type or want to control serialization, you can pass a string with data=json.dumps(payload) and set the header manually. However, the json= parameter is the most direct approach for standard JSON APIs.

Handling JSON Responses and Errors

The resp.json() method parses the response body as JSON. It raises a ContentTypeError if the response does not contain valid JSON or if the Content-Type header is not application/json. To handle non-2xx responses gracefully, check the status code before parsing:

async with session.get(url) as resp:
    if resp.status == 200:
        data = await resp.json()
    else:
        text = await resp.text()
        print(f"Error {resp.status}: {text}")

Alternatively, you can use resp.raise_for_status() to raise an aiohttp.ClientResponseError for any 4xx or 5xx status. This is useful when you want the exception to propagate to a higher-level error handler.

Session Reuse and Connection Pooling

A key advantage of ClientSession is connection reuse. Each session maintains a pool of keep-alive connections, so multiple requests to the same host reuse the same TCP connection. This reduces latency and system resource usage. For example, fetching several resources from the same API:

async with aiohttp.ClientSession() as session:
    for id in [1, 2, 3]:
        async with session.get(f'https://api.example.com/items/{id}') as resp:
            item = await resp.json()
            print(item)

The session also stores cookies, so if you authenticate once, subsequent requests automatically include the session cookie. This behavior makes ClientSession suitable for stateful interactions.

Setting Timeouts and Retry Behavior

Network requests can hang indefinitely if the server is unresponsive. aiohttp provides ClientTimeout to set timeouts for the whole request, connection, or socket read. You can pass a timeout when creating the session or per request:

timeout = aiohttp.ClientTimeout(total=10)  # 10 seconds for the entire request
async with aiohttp.ClientSession(timeout=timeout) as session:
    try:
        async with session.get(url) as resp:
            data = await resp.json()
    except asyncio.TimeoutError:
        print("Request timed out")

aiohttp does not provide built-in retries. If you need retry logic, you must implement it manually, typically by wrapping the request in a loop with exponential backoff. Be careful not to retry on all errors; only retry on transient failures like timeouts or connection errors.

Handling Exceptions and Cancellation

aiohttp raises exceptions derived from aiohttp.ClientError for network problems, and asyncio.TimeoutError for timeouts. A robust client should catch these and decide whether to retry or fail. For example:

try:
    async with session.get(url) as resp:
        resp.raise_for_status()
        data = await resp.json()
except aiohttp.ClientResponseError as e:
    print(f"HTTP error: {e.status}")
except aiohttp.ClientConnectionError:
    print("Connection failed")
except asyncio.TimeoutError:
    print("Request timed out")

When a request is cancelled (e.g., the task is cancelled), aiohttp will clean up the connection. You can use asyncio.shield to protect critical requests from cancellation if needed.

Making Concurrent Requests with asyncio.gather

Because aiohttp is asynchronous, you can issue multiple requests concurrently without threads. The typical pattern is to create a list of coroutine tasks and await them with asyncio.gather:

async def fetch(session, url):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

async def main():
    urls = [
        'https://api.example.com/items/1',
        'https://api.example.com/items/2',
        'https://api.example.com/items/3'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    print(results)

asyncio.run(main())

Using return_exceptions=True prevents one failed request from cancelling the others. This pattern is efficient because the event loop switches between requests while waiting for I/O, allowing many requests to be in flight simultaneously.