# SERP API ใน Python: requests, httpx และ asyncio

> ใช้ SerpCrypto SERP API ใน Python: ไคลเอนต์ requests พร้อม retry, async httpx พร้อมจำกัด concurrency, การแบ่งหน้า และ export CSV โค้ดที่ผ่านการทดสอบ

- URL: https://serpcrypto.com/th/docs/python/
- อัปเดตล่าสุด: 2026-10-22
- ผู้เผยแพร่: SerpCrypto (https://serpcrypto.com)

ไคลเอนต์ที่ใช้ซ้ำได้พร้อม retry, ลูปแบ่งหน้าที่หยุดเมื่อผลการค้นหาหมด, การค้นหาแบบ async ที่จำกัดตาม concurrency limit ของคุณ และ CSV รายวันสำหรับอันดับ โค้ดพร้อมคัดลอกวาง ตรวจสอบ syntax แล้ว ไม่ต้องติดตั้งแพ็กเกจใดๆ ของเรา

**คำตอบสั้น — ฉันจะเรียก SERP API จาก Python ได้อย่างไร**

ส่ง GET request ไปที่ `https://serpcrypto.com/api/v1/search` ด้วย `requests` หรือ `httpx` ใส่ key ใน header `Authorization: Bearer` และคำค้นหาใน `params` จากนั้นอ่าน `response.json()` **SerpCrypto** คิด 1 เครดิตต่อการค้นหาที่สำเร็จ เริ่มต้นที่ US$0.30/1K การค้นหา และการ retry เมื่อเรียกไม่สำเร็จนั้นฟรี จ่ายด้วยคริปโต ไม่ต้อง KYC โค้ดด้านล่างใช้เพียง `requests` หรือ `httpx`

## ตั้งค่า Python โดยสรุป

| คุณลักษณะ | ค่า |
| --- | --- |
| ติดตั้ง | `pip install requests` หรือ `pip install httpx` |
| คีย์ | Environment variable `SERPCRYPTO_API_KEY` |
| Header | `Authorization: Bearer <key>` |
| Timeout | 60 ถึง 90 วินาที: การค้นหาเรียกแบบเรียลไทม์ |
| Retry | เฉพาะ 429, 502 และ 504 พร้อม backoff ข้อผิดพลาดไม่มีค่าใช้จ่าย |
| การค้นหาแบบขนาน | สูงสุด `concurrency_limit` จาก `GET /credits` |

## ฉันจะตั้งค่าไคลเอนต์ Python ที่ใช้ซ้ำได้อย่างไร

สร้าง `requests.Session` หนึ่งตัวโดยใส่ key ใน headers และนโยบาย retry สำหรับข้อผิดพลาดชั่วคราวสามรายการ session จะใช้ connection ซ้ำ ดังนั้นลูปการค้นหาจะใช้เวลาน้อยลงกับ TLS handshake บันทึกไฟล์เป็น `serp_client.py` ตัวอย่างถัดไปจะ import มัน

`Retry` จาก urllib3 จะรอนานขึ้นก่อนพยายามใหม่แต่ละครั้ง และจะเชื่อฟัง header `Retry-After` หากมีการส่งมา มันจะไม่ retry `400`, `401` หรือ `402` ซึ่งต้องแก้ไข ไม่ใช่ลองใหม่

*serp_client.py*

```python
"""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())
```

## ฉันจะเก็บผลลัพธ์มากกว่า 10 รายการได้อย่างไร

ขอหน้าถัดไปทีละหน้า ผลลัพธ์ออร์แกนิกประมาณ 10 รายการต่อหน้า ตามที่ Google ให้บริการตั้งแต่เลิกใช้ num=100 ในเดือนกันยายน 2025; แต่ละหน้าเพิ่มคือหนึ่งเครดิตเพิ่มเติม หยุดทันทีเมื่อ `pagination.next_page` หายไป: ไม่มีอะไรให้ดึงอีกแล้ว และคุณยังเก็บเครดิตไว้ได้

สำหรับงานติดตามอันดับส่วนใหญ่ หน้าแรกหรือสามหน้าแรกก็เพียงพอ เพราะคลิกกระจุกตัวอยู่ตรงนั้น สามสิบผลลัพธ์สำหรับหนึ่งคำค้นหาคิดสามเครดิต

*การแบ่งหน้า: สูงสุดสามหน้า หยุดก่อนกำหนด*

```python
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")
```

## ฉันจะรันการค้นหาจำนวนมากแบบขนานได้อย่างไร

ใช้ `httpx.AsyncClient` กับ `asyncio.Semaphore` ที่กำหนดขนาดตาม limit ของบัญชีคุณ สคริปต์อ่าน `concurrency_limit` จาก endpoint เครดิตฟรี ดังนั้นจะปรับตัวได้เมื่อฝ่ายสนับสนุนเพิ่ม limit ของคุณ การค้นหาที่เกิน limit จะได้รับ `429` อยู่แล้ว semaphore ช่วยหลีกเลี่ยงการเสีย attempt เหล่านั้น

แต่ละ task จะ retry ข้อผิดพลาดชั่วคราวด้วยการหยุดพักที่เพิ่มขึ้น และคืนค่า `None` หลังจากล้มเหลวสี่ครั้ง ดังนั้นคำค้นหาที่ดื้อหนึ่งรายการจะไม่บล็อกทั้ง batch

*asyncio + httpx: การค้นหาแบบขนานภายใต้ limit*

```python
"""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"]))
```

## ฉันจะบันทึกอันดับรายวันลงไฟล์ CSV ได้อย่างไร

เพิ่มหนึ่งแถวต่อผลลัพธ์ พร้อมวัน คำค้นหา ตำแหน่ง ลิงก์ และการตั้งค่าที่ API ใช้ การอ่าน `gl` และ `device` จาก `search_parameters` แทนที่จะอ่านจากตัวแปรของคุณเองหมายความว่าไฟล์บันทึกสิ่งที่ใช้จริง

รันสคริปต์วันละครั้งด้วย cron หรือ Windows Task Scheduler สามคำค้นหาบนมือถือคิดสามเครดิตต่อวัน ประมาณ 90 ต่อเดือน

*เพิ่มอันดับวันนี้ลงใน rankings.csv*

```python
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"],
            })
```

## ฉันควรใช้ไคลเอนต์ HTTP ตัวไหนใน Python

ใช้ `requests` สำหรับสคริปต์และ cron job: ใช้งานง่ายและ retry adapter ของมันเสถียรแล้ว ใช้ `httpx` เมื่อคุณต้องการโค้ดแบบ async เช่น web app, agent หรือชุดคำค้นหาหลายร้อยคำ ทั้งสองส่ง request เดียวกันและได้ JSON เดียวกัน API ไม่สนใจว่าคุณเลือกตัวไหน

หลีกเลี่ยง timeout ที่สั้นเกินไปกับทั้งสองไลบรารี client ที่ยอมแพ้หลังห้าวินาทีอาจส่งคำค้นหาที่กำลังทำงานอยู่ซ้ำและจ่ายเงินทั้งสองครั้ง

## สคริปต์ของฉันจะเสียค่าใช้จ่ายเท่าไหร่?

นับจำนวนการค้นหา ไม่ใช่นับผลลัพธ์: หนึ่งการเรียกคือหนึ่งเครดิตเมื่อสำเร็จ งานที่มี 500 คำค้นหา หน้าละหนึ่งหน้า ทุกวันเป็นเวลา 30 วัน คือ 15,000 การค้นหา บนแพ็ก 25,000 การค้นหา (US$18.75) ยังมีเหลือเฟือ และ เครดิตการค้นหาแบบเติมเงินไม่มีวันหมดอายุ: คำค้นหาที่ไม่ได้ใช้จะคงอยู่ในบัญชีของคุณจนกว่าคุณจะใช้

ตรวจสอบยอดคงเหลือจากโค้ดด้วยตัวช่วย `credits()` ด้านบน ไม่มีค่าใช้จ่าย

## คำถามเกี่ยวกับ Python

### มีแพ็กเกจ SerpCrypto บน PyPI ไหม?

ไม่มี ตัวอย่างเรียก endpoint HTTPS ด้วย `requests` หรือ `httpx` ซึ่งเป็นไลบรารีที่ใช้กันแพร่หลายสองตัว จึงไม่มีอะไรของเราที่ต้องติดตั้งหรืออัปเดต client ด้านบนมีประมาณ 40 บรรทัดและคุณเป็นเจ้าของมันเอง

### โค้ดนี้รองรับ Python เวอร์ชันใดบ้าง?

Python 3.9 ขึ้นไป ตัวอย่าง async ใช้ `asyncio.run` ซึ่งมีมาตั้งแต่ Python 3.7 ส่วนที่เหลือใช้เฉพาะ standard library บวกกับ `requests` หรือ `httpx`

### ใช้ API จาก Jupyter notebook ได้ไหม?

ได้ ตั้งค่า `SERPCRYPTO_API_KEY` ก่อนเปิด Jupyter หรืออ่าน API key ด้วย `getpass` ภายใน notebook เพื่อไม่ให้มันไปอยู่ในเซลล์ที่บันทึกไว้ ใน notebook ให้ใช้ `await main(...)` แทน `asyncio.run(main(...))`

### ฉันจะแปลงผลลัพธ์เป็น pandas DataFrame ได้อย่างไร?

ส่งลิสต์ของ dictionary ผลลัพธ์ให้ `pandas.DataFrame(data["organic_results"])` แต่ละ field กลายเป็นคอลัมน์ field ที่เป็น optional และไม่มีค่า เช่น `date` จะกลายเป็นเซลล์ว่าง

---

SerpCrypto: SerpCrypto คือ SERP API สำหรับนักพัฒนาและ AI agent ที่ส่งคืนผลการค้นหา Google และ Bing เป็น JSON จ่ายด้วยคริปโตโดยไม่ต้องอัปโหลดบัตรประชาชน เครดิตการค้นหาแบบเติมเงินจาก US$0.30/1K การค้นหา ไม่มีวันหมดอายุ ไม่ต้อง KYC — ไม่ต้องใช้บัตรประชาชน ไม่ต้องเซลฟี่ ไม่ต้องใช้เอกสาร จ่ายด้วยคริปโต
