How to Build REST Client Libraries -- Step-by-Step Tutorial

Building a REST client library is one of the most practical skills a developer can learn. Whether you are wrapping a third-party API for your team, open-sourcing a SDK, or creating a reusable search client for tools like SearchHive, a well-structured REST client saves time and reduces errors across every project that consumes the API.

This tutorial walks through building a production-ready REST client library in Python, complete with error handling, pagination, async support, and a real-world SearchHive SwiftSearch integration example.

Prerequisites

Before starting, make sure you have:

• Python 3.9+ installed
• httpx for HTTP requests (supports sync and async)
• pydantic for data validation and typed responses
• An API key from the service you want to wrap (we will use SearchHive for examples)

pip install httpx pydantic python-dotenv

Step 1: Define Your Client Architecture

A good REST client separates concerns into layers:

• Transport layer -- HTTP requests, retries, timeouts
• Resource layer -- API-specific endpoints (search, scrape, etc.)
• Model layer -- Typed response objects
• Auth layer -- API key management

Create a project structure:

my_client/
    __init__.py
    client.py      # Main client class
    models.py      # Response models
    exceptions.py  # Custom errors
    resources/
        __init__.py
        search.py  # Search endpoints

Step 2: Set Up the HTTP Transport Layer

The transport layer handles all HTTP communication. Using httpx gives you sync/async support out of the box.

# client.py
import httpx
from typing import Optional, Dict, Any

class BaseClient:
    BASE_URL = "https://api.searchhive.dev/v1"

    def __init__(self, api_key: str, timeout: int = 30):
        self._client = httpx.Client(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
                "User-Agent": "my-client/1.0",
            },
            timeout=timeout,
        )
        self._api_key = api_key

    def _request(
        self, method: str, path: str, **kwargs: Any
    ) -> Dict[str, Any]:
        response = self._client.request(method, path, **kwargs)
        response.raise_for_status()
        return response.json()

    def get(self, path: str, params: Optional[dict] = None):
        return self._request("GET", path, params=params)

    def post(self, path: str, json: Optional[dict] = None):
        return self._request("POST", path, json=json)

    def close(self):
        self._client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

Key design decisions:

• raise_for_status() converts HTTP errors into exceptions immediately
• Context manager (__enter__/__exit__) ensures connections get closed
• Base URL is set once, so resource methods only need the path

Step 3: Create Typed Response Models

Use Pydantic models to validate API responses and give users autocomplete in their IDE.

# models.py
from pydantic import BaseModel, Field
from typing import List, Optional

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str
    position: int

class SearchResponse(BaseModel):
    query: str
    results: List[SearchResult] = Field(default_factory=list)
    total_results: int = 0
    page: int = 1
    credits_remaining: int = 0

class ScrapedPage(BaseModel):
    url: str
    title: str
    content: str
    status_code: int = 200
    links: List[str] = Field(default_factory=list)

Models act as documentation -- anyone using your client can see exactly what fields are available without reading API docs.

Step 4: Build Resource Classes

Resource classes group related endpoints. This pattern mirrors how APIs are structured.

# resources/search.py
from ..models import SearchResponse, ScrapedPage

class SearchResource:
    def __init__(self, client):
        self._client = client

    def web(self, query: str, limit: int = 10, page: int = 1) -> SearchResponse:
        """Run a web search using SwiftSearch."""
        data = self._client.get(
            "/search/web",
            params={"q": query, "limit": limit, "page": page},
        )
        return SearchResponse(**data)

    def news(self, query: str, limit: int = 10) -> SearchResponse:
        """Search news articles."""
        data = self._client.get(
            "/search/news",
            params={"q": query, "limit": limit},
        )
        return SearchResponse(**data)

    def scrape(self, url: str, format: str = "markdown") -> ScrapedPage:
        """Scrape a single URL using ScrapeForge."""
        data = self._client.post(
            "/scrape",
            json={"url": url, "format": format},
        )
        return ScrapedPage(**data)

Step 5: Add Pagination Support

Most APIs paginate results. Build a generic paginator so users do not have to manage page tracking themselves.

# client.py (addition)
from typing import Iterator, TypeVar, Type
from ..models import SearchResponse

T = TypeVar("T")

class PaginatedIterator:
    """Automatically fetches paginated results."""

    def __init__(self, fetch_fn, response_model: Type[T], limit: int = 10):
        self._fetch = fetch_fn
        self._model = response_model
        self._limit = limit
        self._page = 1
        self._buffer = []
        self._exhausted = False

    def __iter__(self):
        return self

    def __next__(self):
        if not self._buffer and not self._exhausted:
            data = self._fetch(page=self._page, limit=self._limit)
            response = self._model(**data)
            self._buffer = response.results
            if len(response.results) < self._limit:
                self._exhausted = True
            self._page += 1

        if self._buffer:
            return self._buffer.pop(0)
        raise StopIteration

Usage:

# Iterate through all search results without managing pages
for result in client.search.web_iter("python web scraping"):
    print(result.title, result.url)

Step 6: Add Async Support

Async clients are essential for high-throughput applications. httpx.AsyncClient makes this straightforward.

# async_client.py
import httpx
import asyncio
from typing import List, Optional

class AsyncBaseClient:
    BASE_URL = "https://api.searchhive.dev/v1"

    def __init__(self, api_key: str, timeout: int = 30):
        self._client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            timeout=timeout,
        )

    async def search_batch(self, queries: List[str]) -> List[dict]:
        """Run multiple searches concurrently."""
        tasks = [
            self._client.get("/search/web", params={"q": q})
            for q in queries
        ]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

    async def close(self):
        await self._client.aclose()

Batch searching is where async shines -- 50 concurrent API calls complete in the time of one.

Step 7: Add Error Handling and Retries

Production clients need robust error handling. Define custom exceptions and add retry logic.

# exceptions.py
class SearchHiveError(Exception):
    """Base exception for the client."""

class AuthenticationError(SearchHiveError):
    """Invalid or missing API key."""

class RateLimitError(SearchHiveError):
    """Too many requests. Retry after the cooldown."""

class NotFoundError(SearchHiveError):
    """Requested resource does not exist."""

# Add retry logic to the transport layer
import time

class BaseClient:
    # ... existing code ...

    def _request_with_retry(
        self,
        method: str,
        path: str,
        max_retries: int = 3,
        backoff_factor: float = 0.5,
        **kwargs,
    ):
        for attempt in range(max_retries):
            try:
                response = self._client.request(method, path, **kwargs)
                if response.status_code == 429:
                    retry_after = float(
                        response.headers.get("Retry-After",
                        backoff_factor * (2 ** attempt))
                    )
                    time.sleep(retry_after)
                    continue
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 401:
                    raise AuthenticationError("Invalid API key") from e
                if e.response.status_code == 404:
                    raise NotFoundError(f"Resource not found: {path}") from e
                if attempt == max_retries - 1:
                    raise SearchHiveError(
                        f"Request failed after {max_retries} retries"
                    ) from e
                time.sleep(backoff_factor * (2 ** attempt))

Step 8: Complete Working Example

Here is a complete, runnable example that uses SearchHive to search and scrape in a single workflow.

import os
from dotenv import load_dotenv
from searchhive import SearchHiveClient
from searchhive.models import SearchResponse, ScrapedPage

load_dotenv()

def main():
    with SearchHiveClient(
        api_key=os.environ["SEARCHHIVE_API_KEY"]
    ) as client:
        # Step 1: Search for relevant pages
        print("Searching...")
        search_response = client.search.web(
            query="best Python REST client libraries 2025",
            limit=5,
        )
        print(f"Found {search_response.total_results} results")

        # Step 2: Scrape the top 3 results
        print("Scraping top results...")
        urls = [r.url for r in search_response.results[:3]]
        scraped = []
        for url in urls:
            page = client.search.scrape(url, format="markdown")
            scraped.append(page)
            print(f"  Scraped: {page.title} ({len(page.content)} chars)")

        # Step 3: Deep dive into scraped content
        print("Analyzing scraped content...")
        for page in scraped:
            deep_results = client.deep_dive.query(
                question="What REST patterns does this recommend?",
                context=page.content[:2000],
            )
            print(f"  {page.title}: {deep_results.answer[:200]}")

if __name__ == "__main__":
    main()

Common Issues and Fixes

Connection timeouts on large payloads: Increase the timeout value when initializing the client. For scraping operations, 60 seconds is a safe default.

Authentication errors: Double-check that your API key is loaded from environment variables, not hardcoded. Verify the key has the correct permissions for the endpoints you are accessing.

Rate limiting: Implement exponential backoff (shown in Step 7). SearchHive includes Retry-After headers -- always respect them rather than using a fixed delay.

Response model validation errors: If the API changes its response format, Pydantic will raise a ValidationError. Log the raw response body in your exception handler to debug schema changes quickly.

SSL errors behind corporate proxies: Pass a custom httpx.Client(verify=False) in development only. In production, install the proxy CA certificate properly.

Next Steps

Now that you have a solid REST client library, consider these enhancements:

• Add a CLI interface using click or typer so users can interact with the API from the terminal
• Publish to PyPI so other teams can install it with pip install my-client
• Add unit tests with pytest and respx (httpx mocking library) for reliable tests without real API calls
• Add logging using Python's logging module at different verbosity levels
• Support streaming responses for APIs that return data incrementally

For a real-world example of a production REST client, check out SearchHive's Python SDK. The free tier gives you 500 API credits to experiment with while building your client library.