Parse documents and get results

This guide builds the smallest useful Pathway integration in Python. It submits a deal package, waits for the asynchronous parse, and retrieves bank analytics. Use a webhook when your application already has a public callback endpoint. Use polling for local scripts, jobs, and the first version of an integration.

Install the client

python -m pip install requests

Set the token in your environment:

export PATHWAY_API_TOKEN="pat_your_token_here"

The examples use this shared setup:

import os
from pathlib import Path

import requests

API_BASE = "https://api.lendpathway.com/api"
TOKEN = os.environ["PATHWAY_API_TOKEN"]

session = requests.Session()
session.headers["Authorization"] = f"Bearer {TOKEN}"

Submit the deal

Send every file for one deal in the same multipart request.

def submit_book(
    paths: list[Path],
    book_name: str,
    webhook_url: str | None = None,
) -> str:
    opened = [path.open("rb") for path in paths]

    try:
        files = [
            ("files", (path.name, handle, "application/pdf"))
            for path, handle in zip(paths, opened)
        ]
        data = {"book_name": book_name}
        if webhook_url:
            data["webhook_url"] = webhook_url

        response = session.post(
            f"{API_BASE}/submit-book",
            files=files,
            data=data,
            timeout=120,
        )
        response.raise_for_status()
        return response.json()["book_id"]
    finally:
        for handle in opened:
            handle.close()


book_id = submit_book(
    paths=[
        Path("jan_statement.pdf"),
        Path("feb_statement.pdf"),
        Path("credit_report.pdf"),
        Path("application.pdf"),
    ],
    book_name="Smith Auto Body",
)

print("Submitted", book_id)

The response only confirms that Pathway accepted the asynchronous workflow. The new Book continues through upload, classification, and parsing after the request closes.

Wait with polling

Poll the Book every few seconds. Stop when the status reaches completed, failed, or cancelled, and put a timeout around the whole wait.

import time


class ParseFailed(RuntimeError):
    pass


def wait_for_book(
    book_id: str,
    *,
    poll_every: float = 5,
    timeout: float = 15 * 60,
) -> dict:
    deadline = time.monotonic() + timeout

    while time.monotonic() < deadline:
        response = session.get(
            f"{API_BASE}/books/{book_id}",
            timeout=30,
        )
        response.raise_for_status()
        book = response.json()

        status = book["parse_status"]
        message = book.get("parse_status_message") or ""
        print(f"[{status}] {message}")

        if status == "completed":
            return book

        if status in {"failed", "cancelled"}:
            raise ParseFailed(f"{status}: {message}")

        time.sleep(poll_every)

    raise TimeoutError(f"Book {book_id} did not finish within {timeout} seconds")


book = wait_for_book(book_id)

Five seconds is a reasonable polling interval. Faster polling rarely makes the result arrive meaningfully sooner and uses the PAT’s request budget.

Fetch the result

After a completed parse, request the output your workflow needs.

def get_json(path: str) -> dict | list:
    response = session.get(f"{API_BASE}{path}", timeout=60)
    response.raise_for_status()
    return response.json()


analytics = get_json(f"/books/{book_id}/analytics")

print("True revenue:", analytics["true_revenue"])
print("Average daily balance:", analytics["average_daily_balance"])
print("Debt-to-income:", analytics["debt_to_income_ratio"])
print("Positions:", len(analytics["positions"]))

The useful result endpoints are:

Endpoint	Content
`GET /books/{id}`	Status and raw parser output for credit, tax, application, identity, and bank documents
`GET /books/{id}/analytics`	Computed bank underwriting data
`GET /books/{id}/statements`	Account-first statement history and daily balances
`GET /books/{id}/tax-analytics`	Tax qualifying-income analysis
`GET /books/{id}/csv-export`	Bank underwriting CSV
`GET /books/{id}/spreadsheet-export`	Generated Excel or PDF from the gallery

Some Books contain no bank statements. In that case /analytics returns 400, while the relevant raw credit, tax, application, or identity result can still be present in book_meta.

Wait with a webhook

A webhook removes the polling loop. Add webhook_url to the submit request and expose a callback that accepts Pathway’s completion payload.

book_id = submit_book(
    paths=[Path("statement.pdf"), Path("application.pdf")],
    book_name="Smith Auto Body",
    webhook_url="https://example.com/webhooks/pathway",
)

The receiver can remain small. Confirm the stored status with an authenticated API request before starting downstream work.

import os
from typing import Literal

import requests
from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel

app = FastAPI()
API_BASE = "https://api.lendpathway.com/api"
TOKEN = os.environ["PATHWAY_API_TOKEN"]


class ParseWebhook(BaseModel):
    book_id: str
    status: Literal["completed", "failed", "cancelled"]
    book_url: str
    error: str | None = None


def process_completed_book(book_id: str) -> None:
    headers = {"Authorization": f"Bearer {TOKEN}"}

    book_response = requests.get(
        f"{API_BASE}/books/{book_id}",
        headers=headers,
        timeout=30,
    )
    book_response.raise_for_status()
    book = book_response.json()

    if book["parse_status"] != "completed":
        return

    analytics_response = requests.get(
        f"{API_BASE}/books/{book_id}/analytics",
        headers=headers,
        timeout=60,
    )

    if analytics_response.status_code == 400:
        # A completed Book can contain only credit, tax, or application data.
        return

    analytics_response.raise_for_status()
    analytics = analytics_response.json()

    # Persist analytics or enqueue the next step in your own system.
    print(book["name"], analytics["true_revenue"])


@app.post("/webhooks/pathway", status_code=202)
def receive_pathway_webhook(
    payload: ParseWebhook,
    background_tasks: BackgroundTasks,
):
    if payload.status == "completed":
        background_tasks.add_task(process_completed_book, payload.book_id)

    return {"accepted": True}

Pathway currently sends one unsigned webhook delivery. Keep the receiver idempotent and use the authenticated Book response as the final status.

Complete polling script

This version can be copied into one file and run directly.

import os
import time
from pathlib import Path

import requests

API_BASE = "https://api.lendpathway.com/api"
TOKEN = os.environ["PATHWAY_API_TOKEN"]

session = requests.Session()
session.headers["Authorization"] = f"Bearer {TOKEN}"


def submit_book(paths: list[Path], name: str) -> str:
    opened = [path.open("rb") for path in paths]
    try:
        response = session.post(
            f"{API_BASE}/submit-book",
            files=[
                ("files", (path.name, handle, "application/pdf"))
                for path, handle in zip(paths, opened)
            ],
            data={"book_name": name},
            timeout=120,
        )
        response.raise_for_status()
        return response.json()["book_id"]
    finally:
        for handle in opened:
            handle.close()


def wait_for_book(book_id: str, timeout: float = 900) -> dict:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        response = session.get(f"{API_BASE}/books/{book_id}", timeout=30)
        response.raise_for_status()
        book = response.json()

        print(book["parse_status"], book.get("parse_status_message") or "")

        if book["parse_status"] == "completed":
            return book
        if book["parse_status"] in {"failed", "cancelled"}:
            raise RuntimeError(book.get("parse_status_message") or book["parse_status"])

        time.sleep(5)

    raise TimeoutError("Pathway parse timed out")


book_id = submit_book(
    [Path("jan.pdf"), Path("feb.pdf"), Path("application.pdf")],
    "Smith Auto Body",
)
wait_for_book(book_id)

response = session.get(f"{API_BASE}/books/{book_id}/analytics", timeout=60)
response.raise_for_status()
analytics = response.json()

print({
    "book_id": book_id,
    "true_revenue": analytics["true_revenue"],
    "average_daily_balance": analytics["average_daily_balance"],
    "debt_to_income_ratio": analytics["debt_to_income_ratio"],
    "positions": len(analytics["positions"]),
})

Continue with Working with analytics data once this flow is returning completed Books.

Overview

Endpoints

Cookbook

Parse documents and get results

Install the client

Submit the deal

Wait with polling

Fetch the result

Wait with a webhook

Complete polling script

​Install the client

​Submit the deal

​Wait with polling

​Fetch the result

​Wait with a webhook

​Complete polling script

Install the client

Submit the deal

Wait with polling

Fetch the result

Wait with a webhook

Complete polling script