Week 10 — Advanced Capstone: Async REST Client

Combine the full advanced track into a production-grade async REST API client library, complete with type hints, retry logic, caching, pytest test suite, and a pip-installable package.

capstoneasyncioaiohttptypingpytestpackage

Duration

⏱ 4 hours

Level

📊 Advanced

Prerequisite

🎯 Advanced Weeks 1–9

OUTCOME

A fully typed, tested, and packaged async HTTP client library

What you'll learn

1Design a clean async client API with type hints
2Implement retry, timeout, and rate-limiting middleware
3Cache responses with TTL using asyncio
4Write 90%+ coverage tests with pytest-asyncio
5Package and install as a distributable library

1. Project Specification

python

# Usage example
from apiclient import AsyncClient, RetryPolicy

async def main():
    policy = RetryPolicy(max_retries=3, backoff=2.0)
    async with AsyncClient("https://api.example.com", retry=policy) as client:
        users = await client.get("/users", params={"page": 1})
        post  = await client.post("/posts", json={"title": "Hello"})
        print(users, post)

2. Architecture

AsyncClient — main client class with session management
RetryPolicy dataclass — max_retries, backoff, retry_on (status codes)
ResponseCache — asyncio-based TTL cache
Middleware protocol — pluggable request/response hooks
Exceptions — APIError, RateLimitError, TimeoutError

3. Key Implementation Points

python

import asyncio, aiohttp
from dataclasses import dataclass, field
from typing import Any, Optional

@dataclass
class RetryPolicy:
    max_retries: int = 3
    backoff: float = 1.0
    retry_on: tuple = (429, 500, 502, 503)

async def _request_with_retry(
    session: aiohttp.ClientSession,
    method: str,
    url: str,
    policy: RetryPolicy,
    **kwargs: Any,
) -> aiohttp.ClientResponse:
    last_exc: Optional[Exception] = None
    for attempt in range(policy.max_retries + 1):
        try:
            resp = await session.request(method, url, **kwargs)
            if resp.status in policy.retry_on and attempt < policy.max_retries:
                await asyncio.sleep(policy.backoff * (2 ** attempt))
                continue
            return resp
        except aiohttp.ClientError as e:
            last_exc = e
            if attempt < policy.max_retries:
                await asyncio.sleep(policy.backoff * (2 ** attempt))
    raise last_exc or RuntimeError("Request failed")

📝 Exercises

Try them yourself first, then open the solution to compare.

Exercise 1

Implement the client

Goal: Build the async REST client library described above.

Requirements

AsyncClient as async context manager
RetryPolicy with exponential backoff
Response cache with TTL
pytest-asyncio test suite (90%+ coverage)
Installable with pip (pyproject.toml)

Grading

· Core client works — 30%
· Retry logic correct — 25%
· Cache implementation — 20%
· Tests 90%+ coverage — 25%

Example code / lecture materials

All lecture materials and example code are openly available on GitHub.

View on GitHub ↗