Python Examples

Complete Python examples for the WikiRest API.

Installation

# Using pip
pip install requests

# Or for async support
pip install httpx

Basic usage

Simple search

import requests
import os

API_KEY = os.environ.get("WIKIREST_API_KEY")
BASE_URL = "https://api.wikirest.com/v1"

def search(query: str, limit: int = 10) -> dict:
    """Search Wikipedia for matching content."""
    response = requests.get(
        f"{BASE_URL}/search",
        headers={"X-API-Key": API_KEY},
        params={"q": query, "limit": limit}
    )
    response.raise_for_status()
    return response.json()

# Example
results = search("machine learning", limit=5)
for hit in results["hits"]:
    print(f"📄 {hit['title']}")
    print(f"   {hit['text'][:150]}...")
    print()

Get a specific chunk

def get_chunk(chunk_id: str) -> dict:
    """Get a single chunk by ID."""
    response = requests.get(
        f"{BASE_URL}/chunk/{chunk_id}",
        headers={"X-API-Key": API_KEY}
    )
    response.raise_for_status()
    return response.json()

# Example
chunk = get_chunk("12345_3")
print(f"Title: {chunk['title']}")
print(f"Section: {chunk['section']}")
print(f"Text: {chunk['text']}")

Get full page

def get_page(page_id: int, format: str = "chunks") -> dict:
    """Get a Wikipedia page with all its content."""
    response = requests.get(
        f"{BASE_URL}/page/{page_id}",
        headers={"X-API-Key": API_KEY},
        params={"format": format}
    )
    response.raise_for_status()
    return response.json()

# Get as chunks
page = get_page(12345, format="chunks")
print(f"Title: {page['title']}")
print(f"Total chunks: {page['total_chunks']}")

# Get as concatenated text
page_text = get_page(12345, format="concat")
print(f"Word count: {page_text['word_count']}")

Async with httpx

import httpx
import asyncio
import os

API_KEY = os.environ.get("WIKIREST_API_KEY")
BASE_URL = "https://api.wikirest.com/v1"

async def search_async(query: str, limit: int = 10) -> dict:
    """Async search using httpx."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/search",
            headers={"X-API-Key": API_KEY},
            params={"q": query, "limit": limit}
        )
        response.raise_for_status()
        return response.json()

async def search_multiple(queries: list[str]) -> list[dict]:
    """Search multiple queries concurrently."""
    async with httpx.AsyncClient() as client:
        tasks = [
            client.get(
                f"{BASE_URL}/search",
                headers={"X-API-Key": API_KEY},
                params={"q": q, "limit": 5}
            )
            for q in queries
        ]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

# Example: concurrent searches
async def main():
    queries = ["quantum computing", "machine learning", "neural networks"]
    results = await search_multiple(queries)

    for query, result in zip(queries, results):
        print(f"\n{query}: {len(result['hits'])} results")

asyncio.run(main())

Reusable client class

import requests
from typing import Optional
from dataclasses import dataclass

@dataclass
class WikiRestClient:
    """WikiRest API client."""
    api_key: str
    base_url: str = "https://api.wikirest.com/v1"

    def _request(self, method: str, endpoint: str, **kwargs) -> dict:
        """Make an API request."""
        headers = kwargs.pop("headers", {})
        headers["X-API-Key"] = self.api_key

        response = requests.request(
            method,
            f"{self.base_url}{endpoint}",
            headers=headers,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

    def search(
        self,
        query: str,
        limit: int = 10,
        offset: int = 0,
        page_id: Optional[int] = None
    ) -> dict:
        """Search Wikipedia content."""
        params = {"q": query, "limit": limit, "offset": offset}
        if page_id:
            params["page_id"] = page_id
        return self._request("GET", "/search", params=params)

    def get_chunk(self, chunk_id: str) -> dict:
        """Get a single chunk."""
        return self._request("GET", f"/chunk/{chunk_id}")

    def get_page(
        self,
        page_id: int,
        format: str = "chunks",
        max_chunks: Optional[int] = None
    ) -> dict:
        """Get a Wikipedia page."""
        params = {"format": format}
        if max_chunks:
            params["max_chunks"] = max_chunks
        return self._request("GET", f"/page/{page_id}", params=params)

    def get_changes(self, since: Optional[str] = None, limit: int = 100) -> dict:
        """Get recently changed pages."""
        params = {"limit": limit}
        if since:
            params["since"] = since
        return self._request("GET", "/changes", params=params)

# Usage
client = WikiRestClient(api_key=os.environ["WIKIREST_API_KEY"])

results = client.search("artificial intelligence", limit=5)
page = client.get_page(12345)

RAG with OpenAI

import os
import requests
import openai

WIKIREST_KEY = os.environ["WIKIREST_API_KEY"]
OPENAI_KEY = os.environ["OPENAI_API_KEY"]

openai.api_key = OPENAI_KEY

def answer_with_wikipedia(question: str, num_sources: int = 3) -> str:
    """Answer a question using Wikipedia as context."""

    # 1. Search Wikipedia for relevant content
    search_response = requests.get(
        "https://api.wikirest.com/v1/search",
        headers={"X-API-Key": WIKIREST_KEY},
        params={"q": question, "limit": num_sources}
    ).json()

    # 2. Build context from results
    sources = []
    for hit in search_response["hits"]:
        sources.append({
            "title": hit["title"],
            "url": hit["url"],
            "text": hit["text"]
        })

    context = "\n\n---\n\n".join([
        f"Source: {s['title']}\nURL: {s['url']}\n\n{s['text']}"
        for s in sources
    ])

    # 3. Generate answer with OpenAI
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": f"""You are a helpful assistant that answers questions
using Wikipedia as your source. Always cite your sources with the article title.

Wikipedia Context:
{context}"""
            },
            {"role": "user", "content": question}
        ],
        temperature=0.7
    )

    answer = response.choices[0].message.content

    # 4. Append source citations
    answer += "\n\n**Sources:**\n"
    for s in sources:
        answer += f"- [{s['title']}]({s['url']})\n"

    return answer

# Example
question = "What are the main principles of quantum computing?"
answer = answer_with_wikipedia(question)
print(answer)

Error handling

import requests
import time
from requests.exceptions import HTTPError

def search_with_retry(query: str, max_retries: int = 3) -> dict:
    """Search with automatic retry on rate limit or server error."""
    last_error = None

    for attempt in range(max_retries):
        try:
            response = requests.get(
                "https://api.wikirest.com/v1/search",
                headers={"X-API-Key": API_KEY},
                params={"q": query, "limit": 10},
                timeout=30
            )
            response.raise_for_status()
            return response.json()

        except HTTPError as e:
            last_error = e
            status = e.response.status_code

            if status == 429:
                # Rate limited
                retry_after = int(e.response.headers.get("Retry-After", 60))
                wait = min(retry_after, 2 ** attempt * 10)
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)

            elif status >= 500:
                # Server error - retry with backoff
                wait = 2 ** attempt
                print(f"Server error. Retrying in {wait}s...")
                time.sleep(wait)

            elif status == 401:
                raise ValueError("Invalid API key") from e

            elif status == 403:
                raise ValueError("API key revoked or expired") from e

            else:
                raise

        except requests.exceptions.Timeout:
            last_error = TimeoutError("Request timed out")
            wait = 2 ** attempt
            print(f"Timeout. Retrying in {wait}s...")
            time.sleep(wait)

    raise Exception(f"Max retries exceeded: {last_error}") from last_error

Response caching

from functools import lru_cache
import hashlib
import json

@lru_cache(maxsize=1000)
def cached_search(query: str, limit: int = 10) -> str:
    """Search with in-memory caching (returns JSON string for hashability)."""
    response = requests.get(
        "https://api.wikirest.com/v1/search",
        headers={"X-API-Key": API_KEY},
        params={"q": query, "limit": limit}
    )
    response.raise_for_status()
    return response.text

def search(query: str, limit: int = 10) -> dict:
    """Search with caching."""
    return json.loads(cached_search(query, limit))

# First call hits API
results = search("test query")

# Second call uses cache
results = search("test query")  # Instant!

# Check cache stats
print(cached_search.cache_info())

Command-line tool

#!/usr/bin/env python3
"""Simple CLI for WikiRest API."""

import argparse
import os
import json
import requests

API_KEY = os.environ.get("WIKIREST_API_KEY")
BASE_URL = "https://api.wikirest.com/v1"

def search(args):
    response = requests.get(
        f"{BASE_URL}/search",
        headers={"X-API-Key": API_KEY},
        params={"q": args.query, "limit": args.limit}
    )
    response.raise_for_status()
    data = response.json()

    if args.json:
        print(json.dumps(data, indent=2))
    else:
        for hit in data["hits"]:
            print(f"\n{'='*60}")
            print(f"📄 {hit['title']}")
            print(f"🔗 {hit['url']}")
            print(f"\n{hit['text']}")

def main():
    parser = argparse.ArgumentParser(description="WikiRest CLI")
    subparsers = parser.add_subparsers()

    # Search command
    search_parser = subparsers.add_parser("search", help="Search Wikipedia")
    search_parser.add_argument("query", help="Search query")
    search_parser.add_argument("-l", "--limit", type=int, default=5)
    search_parser.add_argument("--json", action="store_true")
    search_parser.set_defaults(func=search)

    args = parser.parse_args()
    if hasattr(args, "func"):
        args.func(args)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()