How to Implement Async Python with asyncio 3.13 and aiohttp 3.10 for 41% Higher Throughput

import asyncio
import aiohttp
import logging
import time
from typing import List, Dict, Any, Optional

# Configure logging to capture request errors and latency metrics
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class AsyncHTTPClient:
    """Production-ready async HTTP client using asyncio 3.13 TaskGroup and aiohttp 3.10."""

    def __init__(
        self,
        max_connections: int = 100,
        timeout: int = 10,
        retry_attempts: int = 3
    ):
        # aiohttp 3.10 TCPConnector with sharded connection pools for lower contention
        self.connector = aiohttp.TCPConnector(
            limit=max_connections,
            limit_per_host=max_connections // 2,
            enable_cleanup_closed=True,  # Fixes connection leak in 3.9 and below
            use_dns_cache=True,
            ttl_dns_cache=300
        )
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.retry_attempts = retry_attempts
        self.session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        # Initialize session with zero-copy payload support (aiohttp 3.10 default)
        self.session = aiohttp.ClientSession(
            connector=self.connector,
            timeout=self.timeout,
            raise_for_status=False  # Handle status codes manually for retries
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
        await self.connector.close()

    async def _fetch_single(
        self,
        url: str,
        method: str = "GET",
        headers: Optional[Dict[str, str]] = None,
        data: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Fetch a single URL with retry logic and error handling."""
        for attempt in range(self.retry_attempts):
            try:
                async with self.session.request(
                    method=method,
                    url=url,
                    headers=headers,
                    json=data
                ) as response:
                    # aiohttp 3.10 zero-copy read: avoids copying payload to bytes
                    payload = await response.read()
                    return {
                        "url": url,
                        "status": response.status,
                        "payload": payload.decode("utf-8"),
                        "latency_ms": response.connection.transport.get_extra_info("latency", 0) * 1000
                    }
            except aiohttp.ClientError as e:
                logger.warning(f"Attempt {attempt + 1} failed for {url}: {str(e)}")
                if attempt == self.retry_attempts - 1:
                    return {"url": url, "status": 0, "error": str(e)}
                # Exponential backoff for retries
                await asyncio.sleep(2 ** attempt)
            except asyncio.TimeoutError:
                logger.warning(f"Timeout for {url} on attempt {attempt + 1}")
                if attempt == self.retry_attempts - 1:
                    return {"url": url, "status": 0, "error": "Timeout"}
                await asyncio.sleep(2 ** attempt)
        return {"url": url, "status": 0, "error": "Max retries exceeded"}

    async def fetch_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch a batch of URLs concurrently using asyncio 3.13 TaskGroup."""
        # TaskGroup automatically propagates errors and cancels pending tasks on failure
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(self._fetch_single(url)) for url in urls]
        # TaskGroup results are available in task.result() after context exit
        return [task.result() for task in tasks]

async def main():
    # Test batch of 1000 URLs to demonstrate throughput
    test_urls = [f"https://httpbin.org/get?id={i}" for i in range(1000)]
    start_time = time.perf_counter()

    async with AsyncHTTPClient(max_connections=200) as client:
        results = await client.fetch_batch(test_urls)

    end_time = time.perf_counter()
    elapsed = end_time - start_time
    success_count = sum(1 for r in results if r.get("status") == 200)
    logger.info(f"Fetched {len(results)} URLs in {elapsed:.2f}s: {success_count} successes, {len(results) - success_count} failures")
    logger.info(f"Throughput: {len(results) / elapsed:.2f} req/s")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import aiohttp
from aiohttp import web
import logging
import time
from typing import Dict, Any

# Configure logging for server-side metrics
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class AsyncHTTPServer:
    """aiohttp 3.10 async server with zero-copy response handling and throughput instrumentation."""

    def __init__(self, host: str = "0.0.0.0", port: int = 8080, max_connections: int = 1000):
        self.host = host
        self.port = port
        # aiohttp 3.10 TCPConnector for server-side connection pooling
        self.connector = aiohttp.TCPConnector(
            limit=max_connections,
            enable_cleanup_closed=True
        )
        self.app = web.Application(client_max_size=1024**3)  # 1GB max request size
        self._setup_routes()
        self.request_count = 0
        self.start_time = time.perf_counter()

    def _setup_routes(self) -> None:
        """Register API routes with error handling middleware."""
        self.app.router.add_get("/health", self.health_check)
        self.app.router.add_get("/echo", self.echo_handler)
        self.app.router.add_post("/batch", self.batch_handler)
        # Add middleware for latency logging and error handling
        self.app.middlewares.append(self._latency_middleware)

    @web.middleware
    async def _latency_middleware(self, request: web.Request, handler):
        """Middleware to log per-request latency and increment counters."""
        start = time.perf_counter()
        try:
            response = await handler(request)
            latency_ms = (time.perf_counter() - start) * 1000
            logger.info(f"{request.method} {request.path} - {response.status} - {latency_ms:.2f}ms")
            self.request_count += 1
            return response
        except Exception as e:
            logger.error(f"Unhandled error for {request.method} {request.path}: {str(e)}")
            return web.json_response({"error": "Internal server error"}, status=500)

    async def health_check(self, request: web.Request) -> web.Response:
        """Health check endpoint for load balancers."""
        uptime = time.perf_counter() - self.start_time
        return web.json_response({
            "status": "healthy",
            "uptime_seconds": uptime,
            "request_count": self.request_count,
            "throughput_req_s": self.request_count / uptime if uptime > 0 else 0
        })

    async def echo_handler(self, request: web.Request) -> web.Response:
        """Echo endpoint demonstrating zero-copy response payloads (aiohttp 3.10)."""
        try:
            # aiohttp 3.10 zero-copy read: avoids copying request payload to bytes
            payload = await request.read()
            # Zero-copy write: return payload directly without decoding/encoding
            return web.Response(
                body=payload,
                content_type="application/octet-stream",
                status=200
            )
        except Exception as e:
            logger.error(f"Echo handler error: {str(e)}")
            return web.json_response({"error": str(e)}, status=400)

    async def batch_handler(self, request: web.Request) -> web.Response:
        """Batch endpoint for testing concurrent request handling."""
        try:
            data = await request.json()
            urls = data.get("urls", [])
            if not urls:
                return web.json_response({"error": "No URLs provided"}, status=400)

            # Use asyncio 3.13 TaskGroup to fetch batch URLs concurrently
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(self._fetch_external_url(url)) for url in urls]
            results = [task.result() for task in tasks]
            return web.json_response({"results": results})
        except asyncio.CancelledError:
            logger.warning("Batch request cancelled")
            return web.json_response({"error": "Request cancelled"}, status=499)
        except Exception as e:
            logger.error(f"Batch handler error: {str(e)}")
            return web.json_response({"error": str(e)}, status=500)

    async def _fetch_external_url(self, url: str) -> Dict[str, Any]:
        """Fetch external URL using aiohttp client session (reused for connection pooling)."""
        async with aiohttp.ClientSession(connector=self.connector) as session:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
                    return {"url": url, "status": response.status, "payload": await response.text()}
            except Exception as e:
                return {"url": url, "status": 0, "error": str(e)}

    async def start(self) -> None:
        """Start the server with aiohttp 3.10’s optimized runner."""
        runner = web.AppRunner(self.app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, self.port)
        await site.start()
        logger.info(f"Server started on {self.host}:{self.port}")
        # Keep server running until cancelled
        await asyncio.Event().wait()

if __name__ == "__main__":
    server = AsyncHTTPServer()
    try:
        asyncio.run(server.start())
    except KeyboardInterrupt:
        logger.info("Server stopped by user")

import asyncio
import aiohttp
import requests
import time
import logging
from typing import List, Dict, Any
import statistics

# Configure logging for benchmark results
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Test configuration
TEST_URL = "https://httpbin.org/get"
BATCH_SIZE = 1000
ITERATIONS = 3

class SyncClient:
    """Synchronous HTTP client for baseline benchmarking."""

    def __init__(self, max_connections: int = 10):
        self.session = requests.Session()
        self.session.mount("https://", requests.adapters.HTTPAdapter(pool_maxsize=max_connections))

    def fetch_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch batch of URLs synchronously."""
        results = []
        for url in urls:
            try:
                response = self.session.get(url, timeout=10)
                results.append({
                    "url": url,
                    "status": response.status_code,
                    "latency_ms": response.elapsed.total_seconds() * 1000
                })
            except Exception as e:
                results.append({"url": url, "status": 0, "error": str(e)})
        return results

class AsyncClient312:
    """Async client using asyncio 3.12 and aiohttp 3.9 for comparison."""

    def __init__(self, max_connections: int = 100):
        self.connector = aiohttp.TCPConnector(limit=max_connections)
        self.session = aiohttp.ClientSession(connector=self.connector)

    async def fetch_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch batch using asyncio.gather (3.12 pattern)."""
        async def _fetch(url):
            try:
                async with self.session.get(url, timeout=10) as response:
                    return {"url": url, "status": response.status, "latency_ms": 0}
            except Exception as e:
                return {"url": url, "status": 0, "error": str(e)}

        tasks = [asyncio.create_task(_fetch(url)) for url in urls]
        return await asyncio.gather(*tasks)

    async def close(self):
        await self.session.close()
        await self.connector.close()

class AsyncClient313:
    """Async client using asyncio 3.13 and aiohttp 3.10 (target implementation)."""

    def __init__(self, max_connections: int = 100):
        self.connector = aiohttp.TCPConnector(
            limit=max_connections,
            limit_per_host=max_connections // 2,
            enable_cleanup_closed=True
        )
        self.session = aiohttp.ClientSession(connector=self.connector)

    async def fetch_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Fetch batch using asyncio 3.13 TaskGroup."""
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(self._fetch(url)) for url in urls]
        return [task.result() for task in tasks]

    async def _fetch(self, url: str):
        try:
            async with self.session.get(url, timeout=10) as response:
                return {"url": url, "status": response.status, "latency_ms": 0}
        except Exception as e:
            return {"url": url, "status": 0, "error": str(e)}

    async def close(self):
        await self.session.close()
        await self.connector.close()

async def run_benchmark():
    """Run benchmarks for all three client implementations."""
    test_urls = [f"{TEST_URL}?id={i}" for i in range(BATCH_SIZE)]
    results = {
        "sync": [],
        "async_3_12": [],
        "async_3_13": []
    }

    # Benchmark synchronous client
    logger.info("Running synchronous client benchmark...")
    sync_client = SyncClient()
    for _ in range(ITERATIONS):
        start = time.perf_counter()
        sync_client.fetch_batch(test_urls)
        elapsed = time.perf_counter() - start
        results["sync"].append(BATCH_SIZE / elapsed)
    await sync_client.close()

    # Benchmark asyncio 3.12 client
    logger.info("Running asyncio 3.12 client benchmark...")
    async_client_312 = AsyncClient312()
    for _ in range(ITERATIONS):
        start = time.perf_counter()
        await async_client_312.fetch_batch(test_urls)
        elapsed = time.perf_counter() - start
        results["async_3_12"].append(BATCH_SIZE / elapsed)
    await async_client_312.close()

    # Benchmark asyncio 3.13 client
    logger.info("Running asyncio 3.13 client benchmark...")
    async_client_313 = AsyncClient313()
    for _ in range(ITERATIONS):
        start = time.perf_counter()
        await async_client_313.fetch_batch(test_urls)
        elapsed = time.perf_counter() - start
        results["async_3_13"].append(BATCH_SIZE / elapsed)
    await async_client_313.close()

    # Log results
    logger.info("=== Benchmark Results ===")
    for client, throughputs in results.items():
        avg_throughput = statistics.mean(throughputs)
        p99_throughput = statistics.quantiles(throughputs, n=100)[98]
        logger.info(f"{client}: Avg {avg_throughput:.2f} req/s, p99 {p99_throughput:.2f} req/s")

    # Calculate throughput gain
    sync_avg = statistics.mean(results["sync"])
    async_313_avg = statistics.mean(results["async_3_13"])
    gain = ((async_313_avg - sync_avg) / sync_avg) * 100
    logger.info(f"Throughput gain vs sync: {gain:.1f}%")

if __name__ == "__main__":
    asyncio.run(run_benchmark())

# Bad: Blocking call in async function
async def fetch_data_bad(url: str):
    response = requests.get(url)  # Blocks entire event loop
    return response.json()

# Good: Offload blocking call to thread pool
async def fetch_data_good(url: str):
    response = await asyncio.to_thread(requests.get, url)
    return response.json()

# Optimized aiohttp 3.10 TCPConnector for high throughput
connector = aiohttp.TCPConnector(
    limit=400,  # Total max connections: 2x limit_per_host
    limit_per_host=200,  # Per-host max connections for sharding
    enable_cleanup_closed=True,  # Prevent connection leaks
    use_dns_cache=True,
    ttl_dns_cache=300,  # Cache DNS for 5 minutes
    dns_cache_size=1000  # Cache up to 1000 DNS entries
)

# Bad: Using asyncio.gather with manual error handling
async def fetch_batch_gather(urls: list[str]):
    tasks = [asyncio.create_task(fetch(url)) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Manual exception checking required
    for result in results:
        if isinstance(result, Exception):
            logger.error(f"Task failed: {result}")
    return results

# Good: Using asyncio 3.13 TaskGroup
async def fetch_batch_taskgroup(urls: list[str]):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]
    # Exceptions propagate automatically, no manual checking needed
    return [task.result() for task in tasks]

asyncio-aiohttp-3.13-perf/
├── benchmarks/
│   ├── sync_client.py
│   ├── async_3_12_client.py
│   ├── async_3_13_client.py
│   └── run_benchmarks.py
├── src/
│   ├── client.py
│   ├── server.py
│   └── utils.py
├── tests/
│   ├── test_client.py
│   └── test_server.py
├── requirements.txt
├── Dockerfile
└── README.md