Backlink data in FastAPI. Async. Type-safe.

# main.py — FastAPI app with RankParse integration
import httpx
from fastapi import FastAPI, Query, HTTPException
from functools import lru_cache

app = FastAPI()
RANKPARSE_KEY = "rp_your_key"
BASE = "https://api.rankparse.com/v1"


async def rp_get(path: str, params: dict) -> dict:
    """Async helper for RankParse API calls."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE}{path}",
            headers={"X-API-Key": RANKPARSE_KEY},
            params=params,
            timeout=10.0,
        )
        resp.raise_for_status()
        return resp.json()


@app.get("/domain-info")
async def domain_info(domain: str = Query(..., description="Domain to analyze")):
    """Return authority score and tech stack for a domain."""
    try:
        authority, tech = await asyncio.gather(
            rp_get("/domain-authority", {"domain": domain}),
            rp_get("/tech-stack", {"domain": domain}),
        )
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=502, detail=f"RankParse error: {e.response.status_code}")

    return {
        "domain": domain,
        "authority_score": authority["data"]["authority_score"],
        "referring_domains": authority["data"]["referring_domains"],
        "tech_stack": [t["name"] for t in tech["data"].get("technologies", [])],
    }


@app.get("/backlinks")
async def backlinks(
    domain: str = Query(...),
    limit: int = Query(20, ge=1, le=1000),
):
    """Proxy the RankParse backlinks endpoint."""
    result = await rp_get("/backlinks", {"domain": domain, "limit": limit})
    return result["data"]