httpx ও asyncio দিয়ে পেশাদার API ক্লায়েন্ট

mkdir -p ~/projects/api-clients
cd ~/projects/api-clients
uv init --bare
uv add -q httpx pydantic
uv add -q --dev pytest pytest-asyncio respx
# uv-এর বদলে pip দিয়ে:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q httpx pydantic pytest pytest-asyncio respx
ls

Initialized project `api-clients`
pyproject.toml
uv.lock

import httpx

BASE_URL = "http://localhost:8800"

response = httpx.get(f"{BASE_URL}/v1/courses", params={"level": "advanced"})
print(response.status_code, response.http_version)
print(response.url)
print([course["title"] for course in response.json()["items"]])

200 HTTP/1.1
http://localhost:8800/v1/courses?level=advanced
['RAG Systems', 'AI Agents', 'LLMOps']

import httpx2

print(httpx2.Client is httpx.Client)          # দুটো আলাদা লাইব্রেরি
with httpx2.Client(base_url=BASE_URL) as other:
    print(other.get("/health").json())
    print(other.headers["user-agent"].split("/")[0])

False
{'status': 'ok'}
python-httpx2

def count_connections(client, n):
    opened = []

    def trace(event, info):
        if event == "connection.connect_tcp.complete":
            opened.append(event)

    for _ in range(n):
        client.get(f"{BASE_URL}/health", extensions={"trace": trace})
    return len(opened)

with httpx.Client() as shared:
    print("one shared client:  ", count_connections(shared, 5), "connection(s) for 5 requests")

total = 0
for _ in range(5):
    with httpx.Client() as throwaway:
        total += count_connections(throwaway, 1)
print("a client per request:", total, "connection(s) for 5 requests")

one shared client:   1 connection(s) for 5 requests
a client per request: 5 connection(s) for 5 requests

client = httpx.Client(
    base_url=BASE_URL,
    headers={"User-Agent": "bitbyte-notebook/1.0", "Accept": "application/json"},
    timeout=httpx.Timeout(10.0, connect=3.0),
    limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
)
echo = client.get("/v1/echo", params={"q": "vector databases", "k": 3}).json()
print(echo["query"])
print(echo["headers"])
print(client.timeout)

{'q': 'vector databases', 'k': '3'}
{'user-agent': 'bitbyte-notebook/1.0', 'accept': 'application/json'}
Timeout(connect=3.0, read=10.0, write=10.0, pool=10.0)

httpx.HTTPError
├── httpx.HTTPStatusError        4xx/5xx রেসপন্সে raise_for_status()
└── httpx.RequestError           কোনো কাজের রেসপন্সই আসেনি
    └── httpx.TransportError
        ├── httpx.TimeoutException
        │   ├── ConnectTimeout   ReadTimeout   WriteTimeout   PoolTimeout
        └── httpx.NetworkError
            ├── ConnectError     ReadError     WriteError     CloseError

def try_call(label, func):
    try:
        response = func()
        response.raise_for_status()
        print(f"{label}: {response.status_code}")
    except httpx.HTTPStatusError as exc:
        print(f"{label}: HTTPStatusError {exc.response.status_code} {exc.response.json()['detail']!r}")
    except httpx.TimeoutException as exc:
        print(f"{label}: {type(exc).__name__}")
    except httpx.RequestError as exc:
        print(f"{label}: {type(exc).__name__} {exc}")

try_call("slow", lambda: client.get("/v1/slow", params={"seconds": 2}, timeout=httpx.Timeout(5.0, read=0.5)))
try_call("down", lambda: client.get("http://localhost:8001/health"))
try_call("missing", lambda: client.get("/v1/courses/99"))
try_call("ok", lambda: client.get("/v1/courses/8"))

slow: ReadTimeout
down: ConnectError [Errno 111] Connection refused
missing: HTTPStatusError 404 'Course 99 not found'
ok: 200

# file: retrying.py
"""স্ট্যাটাস দেখে রিট্রাই: এক্সপোনেনশিয়াল ব্যাকঅফ, ফুল জিটার আর Retry-After।"""
import random
import time

import httpx

RETRY_STATUSES = {429, 502, 503, 504}
RETRY_ERRORS = (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout)


def retry_after_seconds(response):
    """সংখ্যায় লেখা Retry-After হেডার থেকে সেকেন্ড, নয়তো None।"""
    value = response.headers.get("Retry-After", "")
    return float(value) if value.isdigit() else None


def request_with_retries(client, method, url, *, attempts=4, base_delay=0.2, max_wait=5.0,
                         log=print, **kwargs):
    for attempt in range(1, attempts + 1):
        response = None
        try:
            response = client.request(method, url, **kwargs)
            if response.status_code not in RETRY_STATUSES:
                return response
            reason = f"HTTP {response.status_code}"
        except RETRY_ERRORS as exc:
            reason = type(exc).__name__
            if attempt == attempts:
                raise
        if attempt == attempts:
            return response
        wait = retry_after_seconds(response) if response is not None else None
        if wait is None:
            wait = random.uniform(0, base_delay * 2 ** attempt)     # "ফুল জিটার"
        if wait > max_wait:
            log(f"  attempt {attempt}: {reason}, server asks for {wait:.0f} s, giving up")
            return response
        log(f"  attempt {attempt}: {reason}, retrying")
        time.sleep(wait)

from retrying import request_with_retries

print("flaky endpoint:")
r = request_with_retries(client, "GET", "/v1/flaky", params={"key": "p07-demo", "fail_times": 2})
print(" ", r.status_code, r.json())

print("rate-limited endpoint:")
for call in range(4):
    r = request_with_retries(client, "GET", "/v1/limited", headers={"X-Client-Id": "p07-notebook"})
    print(" ", r.status_code, r.json(), r.headers.get("Retry-After"))

flaky endpoint:
  attempt 1: HTTP 503, retrying
  attempt 2: HTTP 503, retrying
  200 {'ok': True, 'attempt': 3}
rate-limited endpoint:
  200 {'ok': True, 'remaining': 2} None
  200 {'ok': True, 'remaining': 1} None
  200 {'ok': True, 'remaining': 0} None
  attempt 1: HTTP 429, server asks for 10 s, giving up
  429 {'detail': 'Too many requests'} 10

def iter_courses(client, level=None, per_page=4):
    """next_page None না হওয়া পর্যন্ত পরের পেজে গিয়ে প্রতিটা কোর্স yield করে।"""
    page = 1
    while page is not None:
        params = {"page": page, "per_page": per_page}
        if level:
            params["level"] = level
        response = client.get("/v1/courses", params=params)
        response.raise_for_status()
        body = response.json()
        print(f"  fetched page {body['page']} ({len(body['items'])} items)")
        yield from body["items"]
        page = body["next_page"]

titles = [course["title"] for course in iter_courses(client)]
print(len(titles), "courses; last:", titles[-1])

first_two = []
for course in iter_courses(client, per_page=2):
    first_two.append(course["id"])
    if len(first_two) == 2:
        break
print(first_two)

  fetched page 1 (4 items)
  fetched page 2 (4 items)
  fetched page 3 (2 items)
10 courses; last: LLMOps
  fetched page 1 (2 items)
[1, 2]

# file: bitbyte_client.py
"""BitByte Practice API-র জন্য ছোট একটা টাইপ করা ক্লায়েন্ট।"""
from collections.abc import Iterator

import httpx
from pydantic import BaseModel, ValidationError


class Course(BaseModel):
    id: int
    title: str
    level: str
    hours: int
    rating: float


class CoursePage(BaseModel):
    page: int
    total: int
    next_page: int | None
    items: list[Course]


class Weather(BaseModel):
    city: str
    temp_c: float
    humidity: int
    condition: str


class BitByteError(Exception):
    """এই ক্লায়েন্ট যত এরর তোলে, সবার বেস ক্লাস।"""


class NotFound(BitByteError):
    pass


class UpstreamError(BitByteError):
    """API ব্যর্থ হয়েছে, বা এমন ডেটা দিয়েছে যা কাজে লাগানো যায় না।"""


class BitByteClient:
    def __init__(self, base_url: str, *, timeout: float = 10.0,
                 transport: httpx.BaseTransport | None = None) -> None:
        self._http = httpx.Client(
            base_url=base_url,
            timeout=httpx.Timeout(timeout, connect=3.0),
            headers={"User-Agent": "bitbyte-client/0.1"},
            transport=transport,
        )

    def __enter__(self) -> "BitByteClient":
        return self

    def __exit__(self, *exc_info) -> None:
        self.close()

    def close(self) -> None:
        self._http.close()

    def _get(self, path: str, **params) -> dict:
        try:
            response = self._http.get(path, params=params)
        except httpx.RequestError as exc:
            raise UpstreamError(f"GET {path} failed: {type(exc).__name__}") from exc
        if response.status_code == 404:
            raise NotFound(response.json().get("detail", path))
        if response.is_error:
            raise UpstreamError(f"GET {path} returned {response.status_code}")
        return response.json()

    def get_course(self, course_id: int) -> Course:
        return self._parse(Course, self._get(f"/v1/courses/{course_id}"))

    def list_courses(self, *, level: str | None = None, page: int = 1, per_page: int = 5) -> CoursePage:
        params = {"page": page, "per_page": per_page}
        if level:
            params["level"] = level
        return self._parse(CoursePage, self._get("/v1/courses", **params))

    def iter_courses(self, *, level: str | None = None) -> Iterator[Course]:
        page: int | None = 1
        while page is not None:
            result = self.list_courses(level=level, page=page, per_page=20)
            yield from result.items
            page = result.next_page

    def weather(self, city: str) -> Weather:
        return self._parse(Weather, self._get("/v1/weather", city=city))

    @staticmethod
    def _parse(model, data):
        try:
            return model.model_validate(data)
        except ValidationError as exc:
            raise UpstreamError(f"unexpected {model.__name__} payload: {exc.error_count()} error(s)") from exc

from bitbyte_client import BitByteClient, NotFound

with BitByteClient(BASE_URL) as bitbyte:
    course = bitbyte.get_course(8)
    print(course)
    print(sum(c.hours for c in bitbyte.iter_courses(level="beginner")), "hours of beginner courses")
    print(bitbyte.weather("Sylhet").model_dump())
    try:
        bitbyte.get_course(99)
    except NotFound as exc:
        print("NotFound:", exc)

id=8 title='RAG Systems' level='advanced' hours=10 rating=4.9
42 hours of beginner courses
{'city': 'Sylhet', 'temp_c': 27.4, 'humidity': 85, 'condition': 'Rain'}
NotFound: Course 99 not found

import asyncio
import time


async def fetch_slow(client, seconds):
    response = await client.get("/v1/slow", params={"seconds": seconds})
    return response.json()["waited_seconds"]


async def compare():
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as aclient:
        start = time.perf_counter()
        for _ in range(4):
            await fetch_slow(aclient, 0.5)
        sequential = time.perf_counter() - start

        start = time.perf_counter()
        results = await asyncio.gather(*(fetch_slow(aclient, 0.5) for _ in range(4)))
        concurrent = time.perf_counter() - start
    print("results:", results)
    print("sequential took at least 2 s:", sequential >= 2.0)
    print("concurrent took under 1 s:  ", concurrent < 1.0)

asyncio.run(compare())

results: [0.5, 0.5, 0.5, 0.5]
sequential took at least 2 s: True
concurrent took under 1 s:   True

async def fetch_all(paths, limit=3):
    semaphore = asyncio.Semaphore(limit)
    in_flight = peak = 0

    async def fetch(aclient, path):
        nonlocal in_flight, peak
        async with semaphore:                      # ৩টা চলতে থাকলে এখানে অপেক্ষা করে
            in_flight += 1
            peak = max(peak, in_flight)
            try:
                response = await aclient.get(path)
                response.raise_for_status()
                return response.json()
            finally:
                in_flight -= 1

    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as aclient:
        results = await asyncio.gather(*(fetch(aclient, p) for p in paths), return_exceptions=True)
    return results, peak

cities = ["Dhaka", "Chattogram", "Sylhet", "Khulna", "Rajshahi"]
paths = [f"/v1/weather?city={c}" for c in cities] + [f"/v1/courses/{i}" for i in (1, 2, 3, 6, 7, 9, 10, 99)]

results, peak = asyncio.run(fetch_all(paths))
ok = [r for r in results if not isinstance(r, Exception)]
failed = [(p, type(r).__name__) for p, r in zip(paths, results) if isinstance(r, Exception)]
print(len(ok), "ok, peak concurrency:", peak)
print("failed:", failed)
print([r.get("city") or r.get("title") for r in ok][:6])

12 ok, peak concurrency: 3
failed: [('/v1/courses/99', 'HTTPStatusError')]
['Dhaka', 'Chattogram', 'Sylhet', 'Khulna', 'Rajshahi', 'How AI Works']

async def with_deadline():
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as aclient:
        try:
            async with asyncio.timeout(1.0):
                await asyncio.gather(fetch_slow(aclient, 0.2), fetch_slow(aclient, 3))
        except TimeoutError:
            print("deadline of 1 s exceeded: both requests were cancelled")

asyncio.run(with_deadline())

deadline of 1 s exceeded: both requests were cancelled

import json


async def stream_chat(question):
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=httpx.Timeout(10.0, read=30.0)) as aclient:
        login = await aclient.post("/v1/token", json={"username": "nadia", "password": "s3cret-pass"})
        token = login.json()["access_token"]
        payload = {"messages": [{"role": "user", "content": question}], "stream": True}
        headers = {"Authorization": f"Bearer {token}"}

        async with aclient.stream("POST", "/v1/chat", json=payload, headers=headers) as response:
            response.raise_for_status()
            print(response.headers["content-type"])
            async for line in response.aiter_lines():
                if not line.startswith("data: "):
                    continue                       # ফাঁকা বিভাজক লাইন, কমেন্ট, keep-alive
                event = json.loads(line.removeprefix("data: "))
                if event.get("done"):
                    print("\nusage:", event["usage"])
                else:
                    print(event["delta"], end="", flush=True)

asyncio.run(stream_chat("What is an embedding?"))

text/event-stream; charset=utf-8
This is a practice answer about: What is an embedding. A real model would say more. 
usage: {'prompt_tokens': 4, 'completion_tokens': 16, 'total_tokens': 20}

# file: test_bitbyte_client.py
import httpx
import pytest

from bitbyte_client import BitByteClient, NotFound, UpstreamError

BASE = "https://api.example.test"
COURSE = {"id": 8, "title": "RAG Systems", "level": "advanced", "hours": 10, "rating": 4.9}


@pytest.fixture
def bitbyte():
    with BitByteClient(BASE) as client:
        yield client


def test_get_course_returns_a_model(bitbyte, respx_mock):
    route = respx_mock.get(f"{BASE}/v1/courses/8").respond(200, json=COURSE)
    course = bitbyte.get_course(8)
    assert course.title == "RAG Systems"
    assert route.call_count == 1


def test_404_becomes_not_found(bitbyte, respx_mock):
    respx_mock.get(f"{BASE}/v1/courses/99").respond(404, json={"detail": "Course 99 not found"})
    with pytest.raises(NotFound, match="Course 99"):
        bitbyte.get_course(99)


def test_timeout_becomes_upstream_error(bitbyte, respx_mock):
    respx_mock.get(f"{BASE}/v1/weather").mock(side_effect=httpx.ReadTimeout("slow"))
    with pytest.raises(UpstreamError, match="ReadTimeout"):
        bitbyte.weather("Dhaka")


def test_changed_payload_is_caught(bitbyte, respx_mock):
    broken = {**COURSE, "rating": "excellent"}          # আপস্ট্রিম API একটা টাইপ বদলে ফেলেছে
    respx_mock.get(f"{BASE}/v1/courses/8").respond(200, json=broken)
    with pytest.raises(UpstreamError, match="unexpected Course payload"):
        bitbyte.get_course(8)


def test_query_parameters_are_sent(bitbyte, respx_mock):
    route = respx_mock.get(f"{BASE}/v1/courses", params={"level": "advanced"}).respond(
        200, json={"page": 1, "total": 1, "next_page": None, "items": [COURSE]}
    )
    assert [c.id for c in bitbyte.iter_courses(level="advanced")] == [8]
    assert route.calls.last.request.url.params["per_page"] == "20"


def test_mock_transport_without_respx():
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.headers["user-agent"] == "bitbyte-client/0.1"
        return httpx.Response(200, json={"city": "Dhaka", "temp_c": 30, "humidity": 70, "condition": "Sunny"})

    with BitByteClient(BASE, transport=httpx.MockTransport(handler)) as client:
        assert client.weather("Dhaka").temp_c == 30.0

uv run pytest -q

......                                                                                       [100%]
6 passed in 0.24s

# file: test_async_retry.py
import httpx
import pytest


async def get_with_one_retry(client: httpx.AsyncClient, url: str) -> httpx.Response:
    response = await client.get(url)
    if response.status_code == 503:
        response = await client.get(url)
    return response


@pytest.mark.asyncio
async def test_retries_once_after_503(respx_mock):
    route = respx_mock.get("https://api.example.test/v1/flaky").mock(
        side_effect=[httpx.Response(503), httpx.Response(200, json={"ok": True})]
    )
    async with httpx.AsyncClient() as client:
        response = await get_with_one_retry(client, "https://api.example.test/v1/flaky")
    assert response.json() == {"ok": True}
    assert route.call_count == 2

uv run pytest -q test_async_retry.py

.                                                                                            [100%]
1 passed in 0.02s

SENSITIVE_HEADERS = {"authorization", "x-api-key", "cookie", "set-cookie", "proxy-authorization"}


def mask(value):
    """দুটো কি আলাদা করে চেনার মতো অংশটুকুই রাখে।"""
    return "***" + value[-4:] if len(value) > 8 else "***"


def safe_headers(headers):
    return {k: (mask(v) if k.lower() in SENSITIVE_HEADERS else v) for k, v in headers.items()
            if k.lower() in SENSITIVE_HEADERS | {"content-type", "x-request-id"}}


def log_request(request):
    print("->", request.method, request.url.path, safe_headers(request.headers))


def log_response(response):
    print("<-", response.status_code, response.request.method, response.request.url.path)


with httpx.Client(base_url=BASE_URL, event_hooks={"request": [log_request], "response": [log_response]}) as logged:
    token = logged.post("/v1/token", json={"username": "nadia", "password": "s3cret-pass"}).json()["access_token"]
    logged.get("/v1/me", headers={"Authorization": f"Bearer {token}", "X-Request-ID": "req-42"})
    logged.post("/v1/reviews", headers={"X-API-Key": "practice-key-123"},
                json={"course_id": 8, "rating": 5, "comment": "Clear and practical"})

-> POST /v1/token {'content-type': 'application/json'}
<- 200 POST /v1/token
-> GET /v1/me {'authorization': '***-001', 'x-request-id': 'req-42'}
<- 200 GET /v1/me
-> POST /v1/reviews {'x-api-key': '***-123', 'content-type': 'application/json'}
<- 201 POST /v1/reviews

cd ~/projects
uv init -q --lib --vcs none bitbyte-client
cd bitbyte-client
cp ../api-clients/bitbyte_client.py src/bitbyte_client/client.py
uv add -q httpx pydantic
find . -path ./.venv -prune -o -type f -print | sort

./.python-version
./README.md
./pyproject.toml
./src/bitbyte_client/__init__.py
./src/bitbyte_client/client.py
./src/bitbyte_client/py.typed
./uv.lock

# file: src/bitbyte_client/__init__.py
"""BitByte Practice API-র টাইপ করা ক্লায়েন্ট।"""
from bitbyte_client.client import (
    BitByteClient,
    BitByteError,
    Course,
    CoursePage,
    NotFound,
    UpstreamError,
    Weather,
)

__all__ = ["BitByteClient", "BitByteError", "Course", "CoursePage", "NotFound", "UpstreamError", "Weather"]
__version__ = "0.1.0"

uv build -q
ls dist
uv run python -c "import bitbyte_client; print(bitbyte_client.__version__, bitbyte_client.BitByteClient.__name__)"

bitbyte_client-0.1.0-py3-none-any.whl
bitbyte_client-0.1.0.tar.gz
0.1.0 BitByteClient

# 1. সহজ
with httpx.Client(base_url=BASE_URL, timeout=5.0) as c:
    reports = [c.get("/v1/weather", params={"city": city}).json() for city in cities]
hottest = max(reports, key=lambda r: r["temp_c"])
print(hottest["city"], hottest["temp_c"])


# 2. মাঝারি
async def course_titles(ids, limit):
    semaphore = asyncio.Semaphore(limit)

    async def one(aclient, course_id):
        async with semaphore:
            response = await aclient.get(f"/v1/courses/{course_id}")
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return response.json()

    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as aclient:
        courses = await asyncio.gather(*(one(aclient, i) for i in ids))
    return {c["id"]: c["title"] for c in courses if c is not None}

titles = asyncio.run(course_titles(range(1, 13), limit=4))
print(len(titles), titles[10])

Rajshahi 33.6
10 LLMOps

cd ~/projects/api-clients

# file: test_retrying.py
import httpx

from retrying import request_with_retries


def test_flaky_endpoint_succeeds_on_third_attempt(respx_mock):
    route = respx_mock.get("https://api.example.test/v1/flaky").mock(
        side_effect=[httpx.Response(503), httpx.Response(503), httpx.Response(200, json={"ok": True})]
    )
    with httpx.Client(base_url="https://api.example.test") as client:
        response = request_with_retries(client, "GET", "/v1/flaky", base_delay=0, log=lambda msg: None)
    assert response.status_code == 200
    assert route.call_count == 3

uv run pytest -q test_retrying.py

.                                                                                            [100%]
1 passed in 0.02s