คู่มือนักพัฒนา Python SDK

ภาพรวมโดยสังเขป

NextPDF Python Software Development Kit (SDK) เป็นไคลเอนต์แบบบางที่มีการกำหนดชนิดข้อมูลอย่างชัดเจนสำหรับปลายทาง NextPDF Connect แอปพลิเคชันของคุณรับผิดชอบการตรวจสอบความถูกต้องของอินพุต Portable Document Format (PDF) การจัดการข้อมูลรับรอง และนโยบายการทำงานพร้อมกัน ส่วน SDK รับผิดชอบการสร้างคำขอ การขนส่ง และการกำหนดชนิดข้อมูลของการตอบกลับ รักษาขอบเขตนี้ให้ชัดเจน: อ่าน PDF อย่างปลอดภัย เลือกไคลเอนต์ เรียกเมท็อด ast ที่ต้องการ และจัดการความล้มเหลวเฉพาะกรณี

ใช้คู่มือนี้เมื่อคุณสร้างบริการสกัดข้อมูล งาน batch แบบ asyncio เครื่องมือสำหรับเอเจนต์ปัญญาประดิษฐ์ (AI) หรือเวิร์กโฟลว์บรรทัดคำสั่งรอบ SDK คู่มือนี้ถือว่าคุณได้อ่าน ภาพรวม และ เริ่มต้นฉบับย่อ แล้ว และมี Python 3.10 หรือใหม่กว่าพร้อมปลายทาง NextPDF Connect

ขอบเขตเชิงสถาปัตยกรรม

เลเยอร์	รับผิดชอบโดย	หน้าที่รับผิดชอบ	อย่าใส่ไว้ที่นี่
แหล่งอินพุต	แอปพลิเคชัน	อนุญาตสิทธิ์ผู้เรียก ตรวจสอบความถูกต้องของแหล่ง PDF และเลือกนโยบายการสกัดข้อมูล	ค่าคงที่ Uniform Resource Locator (URL) ของปลายทางหรือข้อมูลรับรอง
การสร้างไคลเอนต์	แอปพลิเคชัน	อ่าน `base_url` และ `api_key` จากสภาพแวดล้อมหรือตัวจัดการความลับ	ความลับที่ฮาร์ดโค้ด
`NextPDF` / `AsyncNextPDF`	SDK	สร้างคำขอ เรียก Connect และส่งคืนโมเดล Pydantic ที่กำหนดชนิดข้อมูลแล้ว	ตรรกะเชิงโดเมนหรือนโยบายการจัดเก็บ
`ast` เนมสเปซเมท็อด	SDK	แมปการเรียกเมท็อดไปยังปลายทาง Connect และแยกวิเคราะห์การตอบกลับ	นโยบายลองใหม่หรือ backoff ที่เกินกว่าค่าที่คุณกำหนดไว้
NextPDF Connect endpoint	การปรับใช้	รันการสกัดข้อมูลและบังคับใช้การยืนยันตัวตน โควตา และสิทธิ์การใช้งาน	การอนุญาตสิทธิ์ของแอปพลิเคชัน

SDK ไม่ทำการรู้จำอักขระด้วยแสง (OCR) หาก PDF เป็นไฟล์สแกนหรือมีเฉพาะรูปภาพ ให้รัน OCR ก่อนการสกัดข้อมูล และถือว่าขั้นตอนนั้นเป็นความรับผิดชอบของแอปพลิเคชันซึ่งอยู่นอกขอบเขตนี้

วงจรชีวิตขณะรันไทม์

ขั้น	พฤติกรรม	การดำเนินการของนักพัฒนา
การสร้างไคลเอนต์	`base_url` และ `api_key` จะถูกตรวจสอบความถูกต้อง หากค่าใดค่าหนึ่งว่างเปล่าจะทำให้เกิด `ValueError`	อ่านทั้งสองค่าจากสภาพแวดล้อม อย่าเขียนค่าเหล่านั้นไว้ในโค้ดโดยตรง
การสร้างแบ็กเอนด์	แบ็กเอนด์ระยะไกลจะเปิด connection pool ไปยัง Connect	ใช้ไคลเอนต์ตัวเดียวซ้ำในการเรียกทั้งหมด แทนการสร้างใหม่ต่อคำขอ
การเรียกเมท็อด	เมท็อด `ast` จะ serialize คำขอ ส่งไบต์ PDF และแยกวิเคราะห์การตอบกลับเป็นโมเดล Pydantic	ส่ง `bytes` ที่ผ่านการตรวจสอบความถูกต้องแล้ว
การแมปข้อผิดพลาด	SDK จะแมปสถานะ Hypertext Transfer Protocol (HTTP) ที่ไม่สำเร็จไปยังคลาสย่อยของข้อยกเว้นเฉพาะกรณี	จับคลาสที่เฉพาะเจาะจงที่สุดก่อน
การปิดระบบ	`AsyncNextPDF.close()` จะปล่อย connection pool ตัวจัดการบริบทแบบ async จะเรียกใช้ให้คุณ	ใช้ `async with` หรือเรียก `close()` ในบล็อก `finally`

โครงสร้างแอปพลิเคชันที่แนะนำ

เส้นทาง	วัตถุประสงค์
`app/pdf/clients.py`	สร้างและแคช `NextPDF` หรือ `AsyncNextPDF` ที่กำหนดค่าแล้ว
`app/pdf/extraction.py`	ตัวห่อหุ้มระดับแอปพลิเคชันรอบการเรียกเมท็อด `ast` ต่าง ๆ
`app/pdf/validation.py`	การตรวจสอบความถูกต้องของแหล่ง PDF การจำกัดขนาด และการตรวจสอบเนื้อหา
`tests/pdf/`	การทดสอบการสกัดข้อมูล โหมดความล้มเหลว และการประมวลผล batch แบบ async

แยกการตรวจสอบความถูกต้องของ PDF ออกจากการสกัดข้อมูล ส่งเฉพาะไบต์ที่ได้รับอนุญาตและผ่านการตรวจสอบขนาดแล้วเข้าสู่เลเยอร์การสกัดข้อมูล และยังต้องพึ่งพาปลายทางเพื่อเสริมการป้องกันเชิงลึก

import os

from nextpdf import NextPDF


def build_client() -> NextPDF:
    """Construct a synchronous client from environment configuration.

    Raises:
        KeyError: When a required environment variable is missing.
    """
    base_url = os.environ["NEXTPDF_BASE_URL"]
    api_key = os.environ["NEXTPDF_API_KEY"]
    return NextPDF(base_url=base_url, api_key=api_key)

รูปแบบไคลเอนต์แบบซิงโครนัส

ใช้ไคลเอนต์ NextPDF แบบซิงโครนัสสำหรับสคริปต์ งาน batch และโน้ตบุ๊ก ตรวจสอบความถูกต้องของอินพุตก่อนเรียก SDK และจัดการเฉพาะความล้มเหลวที่การเรียกนั้นอาจทำให้เกิดขึ้น

from pathlib import Path

from nextpdf import (
    NextPDF,
    CitedTextBlock,
    NextPDFAPIError,
    NextPDFError,
    QuotaExceededError,
)

MAX_PDF_BYTES = 100 * 1024 * 1024  # Reject documents above 100 MiB for the in-memory path.


def read_pdf(path: Path) -> bytes:
    """Read and validate a PDF from disk.

    Raises:
        ValueError: When the file is missing, empty, oversized, or not a PDF.
    """
    if not path.is_file():
        raise ValueError(f"Not a file: {path}")
    data = path.read_bytes()
    if not data:
        raise ValueError("PDF is empty")
    if len(data) > MAX_PDF_BYTES:
        raise ValueError("PDF exceeds the configured size limit; use the CLI streaming path")
    if not data.startswith(b"%PDF-"):
        raise ValueError("File does not look like a PDF")
    return data


def extract_text(client: NextPDF, path: Path) -> list[CitedTextBlock]:
    """Extract cited text blocks, handling the most specific failures first."""
    pdf_bytes = read_pdf(path)
    try:
        return client.ast.extract_cited_text(pdf_bytes)
    except QuotaExceededError as error:
        raise RuntimeError(f"Quota exceeded; retry after {error.retry_after}s") from error
    except NextPDFAPIError as error:
        raise RuntimeError(f"API error {error.status_code}: {error}") from error
    except NextPDFError as error:
        raise RuntimeError(f"SDK error: {error}") from error

รายการผลลัพธ์หนึ่งรายการมีรูปแบบดังนี้:

block = blocks[0]
print(block.text)                  # the extracted text
print(block.citation.page_index)   # 0-based page index
print(block.citation.confidence)   # 0.0 - 1.0

รูปแบบ async และการประมวลผลแบบ batch

ใช้ไคลเอนต์ AsyncNextPDF แบบอะซิงโครนัสภายในรันไทม์แบบ asyncio เช่น FastAPI สร้างไคลเอนต์ตัวเดียวเป็นตัวจัดการบริบทแบบ async แล้วใช้ร่วมกันในการเรียกพร้อมกันหลายรายการ อย่าเปิดไคลเอนต์ใหม่สำหรับแต่ละเอกสาร จำกัดการทำงานพร้อมกันด้วย semaphore เพื่อให้เป็นไปตามโควตาของปลายทาง

import asyncio
import os

from nextpdf import (
    AsyncNextPDF,
    ExtractCitedTablesResponse,
    NextPDFError,
    QuotaExceededError,
)


async def extract_tables_batch(
    pdfs: list[bytes],
    *,
    max_concurrency: int = 4,
) -> list[ExtractCitedTablesResponse | None]:
    """Extract tables from many PDFs concurrently with one shared client.

    Returns one response per input PDF, or None where extraction failed.
    """
    base_url = os.environ["NEXTPDF_BASE_URL"]
    api_key = os.environ["NEXTPDF_API_KEY"]
    semaphore = asyncio.Semaphore(max_concurrency)

    async with AsyncNextPDF(base_url=base_url, api_key=api_key) as client:

        async def one(pdf_bytes: bytes) -> ExtractCitedTablesResponse | None:
            async with semaphore:
                try:
                    return await client.ast.extract_cited_tables(pdf_bytes)
                except QuotaExceededError as error:
                    # Surface the backpressure signal; do not silently drop it.
                    raise RuntimeError(f"Quota exceeded; retry after {error.retry_after}s") from error
                except NextPDFError:
                    return None

        return await asyncio.gather(*(one(pdf) for pdf in pdfs))

อย่าเขียน except ที่ว่างเปล่าเด็ดขาด ให้จัดการความล้มเหลว แปลงเป็นผลลัพธ์ที่กำหนดไว้ หรือโยนซ้ำ

จุดขยาย

จุดขยาย	ใช้สำหรับ	ข้อจำกัด
`AsyncNextPDF(backend=...)`	แทรกแบ็กเอนด์แบบกำหนดเองหรือแบบโลคัลในการทดสอบ	แบ็กเอนด์ต้องเป็นไปตามโปรโตคอล `PdfBackend`
`api_version` อาร์กิวเมนต์	ปักหมุดเวอร์ชันของ Connect application programming interface (API)	ค่าเริ่มต้นเป็น `v1` เปลี่ยนเฉพาะเมื่อปลายทางรองรับเวอร์ชันเป้าหมายเท่านั้น
การกำหนดค่าสภาพแวดล้อม	จัดเตรียม `NEXTPDF_BASE_URL` และ `NEXTPDF_API_KEY` ให้กับ command-line interface (CLI) และเซิร์ฟเวอร์ Model Context Protocol (MCP)	ปฏิบัติต่อคีย์นั้นเสมือนเป็นความลับที่จำกัดขอบเขตไว้เฉพาะเวิร์กโหลด
เซิร์ฟเวอร์ MCP (`python -m nextpdf.mcp`)	เปิดเผยเครื่องมือสกัดข้อมูลให้แก่เอเจนต์ที่รองรับ MCP	ต้องใช้ส่วนเสริม `nextpdf[mcp]` และปลายทางที่ควบคุมได้

เวิร์กโฟลว์การพัฒนา

ติดตั้ง SDK ด้วย pip install nextpdf หรือใช้ pip install nextpdf[mcp] สำหรับเซิร์ฟเวอร์เอเจนต์
อ่าน NEXTPDF_BASE_URL และ NEXTPDF_API_KEY จากสภาพแวดล้อม เพื่อไม่ให้ความลับเข้าสู่ระบบควบคุมซอร์ส
ตรวจสอบความถูกต้องของแหล่ง PDF ทุกแหล่งในเรื่องการมีอยู่ ขนาด และ magic bytes %PDF- ก่อนเรียก SDK
สร้างไคลเอนต์หนึ่งตัวต่อหนึ่งโพรเซสและนำมาใช้ซ้ำ สำหรับ asyncio ให้คงเปิดไว้ด้วย async with
เรียกเมท็อด ast ที่แคบที่สุดสำหรับงานนั้น: extract_cited_text() สำหรับร้อยแก้ว extract_cited_tables() สำหรับตาราง get_document_ast() เฉพาะเมื่อคุณต้องการโครงสร้างทรีทั้งหมดเท่านั้น
จับข้อยกเว้นที่เฉพาะเจาะจงที่สุดที่คุณสามารถจัดการได้ แล้วจึงถอยกลับไปยัง NextPDFError
สำหรับเอกสารที่เกิน 100 MiB ให้ใช้เส้นทางการสตรีมแบบ CLI แทนการสร้างทุกบล็อกขึ้นในหน่วยความจำ
รัน mypy ในโหมด strict และเพิ่มการทดสอบโหมดความล้มเหลวสำหรับแต่ละข้อยกเว้นที่คุณจัดการ

การจัดการความล้มเหลว

ความล้มเหลว	ข้อยกเว้น	การตอบสนองที่แนะนำ
PDF ที่ไม่มีแท็กเมื่อปิดฮิวริสติก	`AstNoStructTreeError` (HTTP 422)	เปิดโหมดฮิวริสติกบนปลายทางหรือจัดเตรียม PDF ที่มีแท็ก
การหมดเวลาการสร้างฝั่งเซิร์ฟเวอร์	`AstBuildTimeoutError` (HTTP 504)	ลดช่วงหน้าแล้วลองใหม่
ต้องมีระดับสิทธิ์การใช้งาน	`NextPDFLicenseError` (HTTP 402)	อัปเกรดสิทธิ์การใช้งานของเซิร์ฟเวอร์หรือถอยกลับไปใช้คุณลักษณะที่ได้รับอนุญาต
การจำกัดอัตราหรือโควตา	`QuotaExceededError` (HTTP 429)	รอ `retry_after` วินาที แล้วลองใหม่พร้อม backoff
ข้อผิดพลาด HTTP อื่น ๆ	`NextPDFAPIError`	ตรวจสอบ `status_code` และ `error_code` บันทึกล็อกและแสดงข้อผิดพลาดที่กำหนดไว้
ข้อผิดพลาด SDK ใด ๆ	`NextPDFError`	การถอยกลับขั้นสุดท้าย อย่าปล่อยให้หลุดออกไปเป็นข้อยกเว้นที่ไม่ถูกจัดการเด็ดขาด

ปลายทางจะรายงานความล้มเหลวด้วยความหมายของสถานะ HTTP ที่สอดคล้องกับ Request for Comments (RFC) 9110 และเนื้อความข้อผิดพลาดที่เครื่องอ่านได้ซึ่งสอดคล้องกับ RFC 9457 ข้อยกเว้นแต่ละรายการจะเก็บรักษา status_code ต้นทางไว้ แมปความล้มเหลวเหล่านั้นไปยังการตอบกลับข้อผิดพลาดของคุณเอง แทนที่จะปล่อยให้รายละเอียดการขนส่งรั่วไหลไปยังผู้เรียก

ค่าเริ่มต้นที่ปลอดภัย

ประเด็น	ค่าเริ่มต้น	เมื่อใดควรแทนที่
เวอร์ชัน API	`v1`	เปลี่ยนเฉพาะเมื่อปลายทางรองรับเวอร์ชันที่ใหม่กว่าเท่านั้น
การยืนยัน Transport Layer Security (TLS)	เปิดใช้งาน ไม่มีการเปิดสวิตช์ที่ไม่ปลอดภัยให้ใช้	อย่าปิดใช้งานสำหรับการรับส่งข้อมูลในระบบจริงเด็ดขาด
ข้อมูลรับรอง	อ่านจากสภาพแวดล้อม ไม่เคยเขียนในโค้ดโดยตรง	ใช้ตัวจัดการความลับในระบบจริง
การจำกัดขนาดในหน่วยความจำ	ปฏิเสธ PDF ที่เกิน 100 MiB บนเส้นทางฝั่งไคลเอนต์	ลดลงสำหรับบริการแบบหลายผู้เช่า ใช้ CLI สำหรับไฟล์ขนาดใหญ่กว่า
การทำงานพร้อมกัน	จำกัดด้วย semaphore ในการประมวลผล batch แบบ async	ปรับให้เข้ากับโควตาของปลายทาง ไม่ใช่จำนวนคอร์ของโฮสต์
การบันทึกล็อก	บันทึกล็อกชื่อไฟล์ ขนาด สถานะ และระยะเวลา	อย่าบันทึกล็อกไบต์ PDF หรือ API key เด็ดขาด

รายการตรวจสอบการทดสอบ

การทดสอบการสร้างยืนยันว่า base_url หรือ api_key ที่ว่างเปล่าจะทำให้เกิด ValueError
การทดสอบการตรวจสอบความถูกต้องครอบคลุมอินพุตที่ขาดหายไป ว่างเปล่า เกินขนาด และที่ไม่ใช่ PDF
การทดสอบการสกัดข้อมูลยืนยันชนิดของโมเดลที่ส่งคืนและ CitationAnchor บนแต่ละบล็อก
การทดสอบโหมดความล้มเหลวครอบคลุม AstNoStructTreeError, AstBuildTimeoutError, NextPDFLicenseError, QuotaExceededError และ NextPDFAPIError
การทดสอบ async ยืนยันว่าไคลเอนต์ทำงานเป็นตัวจัดการบริบทแบบ async with และว่าการทำงานพร้อมกันอยู่ภายในขอบเขตของ semaphore
การทดสอบวงจรชีวิตยืนยันว่า close() ปล่อยการขนส่งและเป็น idempotent
แทรกแบ็กเอนด์จำลองด้วย AsyncNextPDF(backend=...) เพื่อให้การทดสอบรันได้โดยไม่ต้องมีปลายทางที่ใช้งานจริง

ดูเพิ่มเติม

การอ้างอิง Python API — ทุกเมท็อด โมเดล และข้อยกเว้นของไคลเอนต์
Python CLI — การสกัดข้อมูลแบบสตรีมสำหรับเอกสารขนาดใหญ่
เซิร์ฟเวอร์ Python MCP — เครื่องมือสกัดข้อมูลสำหรับเอเจนต์ AI
ภาพรวม Python SDK — ตัวเลือกแบ็กเอนด์และข้อจำกัด