Python httpx and aiohttp: Modern HTTP Clients for API Testing

import httpx
import pytest

# Sync client
def test_get_users_sync():
    with httpx.Client(base_url="https://api.dodatech.com/v1") as client:
        response = client.get("/users")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)
        assert len(data) > 0
        print(f"Users: {len(data)}")

# Async client
@pytest.mark.asyncio
async def test_get_users_async():
    async with httpx.AsyncClient(base_url="https://api.dodatech.com/v1") as client:
        response = await client.get("/users")
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list)

import asyncio
import httpx

async def fetch_user_data(user_id):
    async with httpx.AsyncClient() as client:
        user_resp = await client.get(
            f"https://api.dodatech.com/v1/users/{user_id}"
        )
        orders_resp = await client.get(
            f"https://api.dodatech.com/v1/users/{user_id}/orders"
        )
        return {
            "user": user_resp.json(),
            "orders": orders_resp.json()
        }

async def fetch_multiple_users(user_ids):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user_data(uid)) for uid in user_ids]

    results = [task.result() for task in tasks]
    print(f"Fetched {len(results)} users with orders")
    for r in results:
        print(f"  {r['user']['email']}: {len(r['orders'])} orders")
    return results

# Run
# results = asyncio.run(fetch_multiple_users([1, 2, 3, 4, 5]))
# Expected output:
# Fetched 5 users with orders
#   user1@test.com: 3 orders
#   user2@test.com: 1 orders
#   user3@test.com: 0 orders

import httpx

# HTTP/2 support
client = httpx.Client(http2=True)
response = client.get("https://api.dodatech.com/v1/users")
print(f"HTTP version: {response.http_version}")
# Expected: HTTP/2

# Connection pooling
# httpx reuses connections automatically
# Configure pool size
limits = httpx.Limits(
    max_connections=100,
    max_keepalive_connections=20,
    keepalive_expiry=30.0
)
client = httpx.Client(limits=limits)

# Event hooks for monitoring
def log_request(request):
    print(f"Request: {request.method} {request.url}")

def log_response(response):
    print(f"Response: {response.status_code} ({response.elapsed.total_seconds():.2f}s)")

client = httpx.Client(
    event_hooks={
        "request": [log_request],
        "response": [log_response]
    }
)

# Mock transport for tests
import httpx

def test_with_mock():
    """Use mock transport to avoid real HTTP calls."""
    def handler(request):
        return httpx.Response(
            status_code=200,
            json={"id": 1, "email": "test@test.com"}
        )

    client = httpx.Client(transport=httpx.MockTransport(handler))
    response = client.get("https://api.test.com/users/1")
    assert response.json()["email"] == "test@test.com"

import aiohttp
import asyncio
import pytest

# Basic aiohttp test
@pytest.mark.asyncio
async def test_with_aiohttp():
    async with aiohttp.ClientSession(
        base_url="https://api.dodatech.com/v1"
    ) as session:
        async with session.get("/users") as response:
            assert response.status == 200
            data = await response.json()
            assert len(data) > 0

# Concurrent requests with aiohttp
async def fetch_all_pages(session, base_url, total_pages):
    async def fetch_page(page):
        async with session.get(f"{base_url}/users?page={page}") as resp:
            return await resp.json()

    tasks = [fetch_page(p) for p in range(1, total_pages + 1)]
    results = await asyncio.gather(*tasks)
    all_users = [user for page in results for user in page]
    print(f"Total users across {total_pages} pages: {len(all_users)}")
    return all_users

# async with aiohttp.ClientSession() as session:
#     users = await fetch_all_pages(session, "https://api.dodatech.com/v1", 5)

import asyncio
import httpx

# httpx WebSocket support
async def test_websocket():
    async with httpx.AsyncClient() as client:
        async with client.ws_connect(
            "wss://api.dodatech.com/v1/ws/notifications",
            headers={"Authorization": "Bearer token"}
        ) as websocket:
            # Send message
            await websocket.send_json({"type": "subscribe", "channel": "orders"})

            # Receive messages
            for _ in range(3):
                message = await websocket.receive_json()
                print(f"WS message: {message['type']}")
                assert message["status"] == "ok"

# aiohttp WebSocket
async def test_websocket_aiohttp():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(
            "wss://api.dodatech.com/v1/ws",
            headers={"Authorization": "Bearer token"}
        ) as ws:
            await ws.send_json({"action": "ping"})
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    data = msg.json()
                    print(f"Received: {data}")
                    if data.get("type") == "pong":
                        break