SERP API di Python: requests, httpx, dan asyncio

"""serp_client.py: a small, reusable client for the SerpCrypto SERP API (requests)."""
import os

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

API_URL = "https://serpcrypto.com/api/v1/search"
CREDITS_URL = "https://serpcrypto.com/api/v1/credits"


def make_session():
    session = requests.Session()
    session.headers["Authorization"] = "Bearer " + os.environ["SERPCRYPTO_API_KEY"]
    retry = Retry(
        total=4,
        backoff_factor=1.0,  # waits about 1 s, 2 s, 4 s between attempts
        status_forcelist=(429, 502, 504),
        allowed_methods=("GET",),
        respect_retry_after_header=True,
    )
    session.mount("https://", HTTPAdapter(max_retries=retry))
    return session


SESSION = make_session()


def search(q, **params):
    """Run one search (1 credit if it succeeds) and return the parsed JSON."""
    r = SESSION.get(API_URL, params={"q": q, **params}, timeout=90)
    r.raise_for_status()
    return r.json()


def credits():
    """Free: credits left, used today and the concurrency limit."""
    r = SESSION.get(CREDITS_URL, timeout=30)
    r.raise_for_status()
    return r.json()


if __name__ == "__main__":
    data = search("project management software", gl="us", hl="en")
    for item in data.get("organic_results", []):
        print(item["position"], item["link"])
    print(credits())

from serp_client import search


def top_results(q, pages=3, **params):
    """Collect organic results over several pages: each page is 1 credit."""
    rows = []
    for page in range(1, pages + 1):
        data = search(q, page=page, **params)
        for item in data.get("organic_results", []):
            rows.append({"page": page, "position": item["position"], "title": item["title"], "link": item["link"]})
        if not data.get("pagination", {}).get("next_page"):
            break  # no further page: stop early and save credits
    return rows


rows = top_results("crm for small business", pages=3, gl="gb", hl="en")
print(len(rows), "results")

"""Run many searches in parallel without exceeding the account's concurrency limit."""
import asyncio
import os

import httpx

API_URL = "https://serpcrypto.com/api/v1/search"
CREDITS_URL = "https://serpcrypto.com/api/v1/credits"
HEADERS = {"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]}


async def search_one(client, gate, q):
    async with gate:  # at most N searches at the same time
        for attempt in range(4):
            r = await client.get(API_URL, params={"q": q, "gl": "us", "hl": "en"})
            if r.status_code in (429, 502, 504):
                await asyncio.sleep(2 ** attempt)
                continue
            r.raise_for_status()
            return q, r.json()
        return q, None


async def main(queries):
    async with httpx.AsyncClient(headers=HEADERS, timeout=90) as client:
        limits = (await client.get(CREDITS_URL)).json()
        gate = asyncio.Semaphore(limits["concurrency_limit"])
        results = await asyncio.gather(*(search_one(client, gate, q) for q in queries))
    for q, data in results:
        first = (data or {}).get("organic_results", [{}])[0].get("link")
        print(f"{q!r}: {first}")


asyncio.run(main(["best vpn router", "home espresso machine", "4k monitor for coding"]))

import csv
from datetime import date

from serp_client import search

KEYWORDS = ["standing desk", "ergonomic chair", "monitor arm"]
FIELDS = ["day", "keyword", "position", "title", "link", "gl", "device"]

with open("rankings.csv", "a", newline="", encoding="utf-8") as fh:
    writer = csv.DictWriter(fh, fieldnames=FIELDS)
    if fh.tell() == 0:
        writer.writeheader()
    for kw in KEYWORDS:
        data = search(kw, gl="gb", hl="en", device="mobile")
        applied = data["search_parameters"]
        for item in data.get("organic_results", []):
            writer.writerow({
                "day": date.today().isoformat(),
                "keyword": kw,
                "position": item["position"],
                "title": item["title"],
                "link": item["link"],
                "gl": applied["gl"],
                "device": applied["device"],
            })