Extract text from an Airflow DAG task.

the-problem

A document-ingestion DAG typically reads a batch of files staged in object storage before the pipeline can run any real logic on them. Loading every file into a task's memory and forwarding it as a request body works for small batches. It scales badly, though, and adds a real risk of a task running out of memory on a big file. A worker-heavy DAG also can't hold one task open indefinitely for a slow OCR job. Doing so starves other tasks of a worker slot.

how-to-wire-it-up

There is no txtfetch provider package for Airflow — no operator to install from PyPI, no connection type to register beyond a plain HTTP connection. A task built with SimpleHttpOperator, or a PythonOperator running requests.post directly, reaches the same REST API a curl command would. Point the request at each file's existing object-storage URL with the url query parameter. No task ever loads a full file into memory just to forward it. For a batch with slow or large documents, pass async=true. Then either poll GET /v1/extract/{job_id} from a downstream task, or supply webhook_url and let a separate consumer pick up each result.

• Pass ?url= with each file's existing storage URL, so no DAG task loads a full document into memory first.
• Automatic OCR covers scanned batches the same way as digital-native files, with no separate OCR operator.
• Async mode returns a job_id per document, so one slow OCR job never occupies a worker slot for the whole run.
• Polling GET /v1/extract/{job_id} from a downstream task, or supplying webhook_url, both fit Airflow's own retry and sensor patterns.
• Idempotency-Key support keeps a re-triggered or backfilled DAG run from re-billing extractions it already completed.

pass-a-link-not-a-file

curl -X POST "https://api.txtfetch.com/v1/extract?url=https://example.com/report.pdf" \
  -H "Authorization: Bearer $TXTFETCH_KEY"

import os
import requests

r = requests.post(
    "https://api.txtfetch.com/v1/extract",
    headers={"Authorization": f"Bearer {os.environ['TXTFETCH_KEY']}"},
    params={"url": "https://example.com/report.pdf"},
)

print(r.json()["extracted_text"])

const endpoint = new URL("https://api.txtfetch.com/v1/extract");
endpoint.searchParams.set("url", "https://example.com/report.pdf");

const res = await fetch(endpoint, {
  method: "POST",
  headers: { Authorization: `Bearer ${process.env.TXTFETCH_KEY}` },
});

const { extracted_text } = await res.json();
console.log(extracted_text);

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
)

type extractResponse struct {
	Status        string `json:"status"`
	ExtractedText string `json:"extracted_text"`
}

func main() {
	endpoint, err := url.Parse("https://api.txtfetch.com/v1/extract")
	if err != nil {
		panic(err)
	}
	q := endpoint.Query()
	q.Set("url", "https://example.com/report.pdf")
	endpoint.RawQuery = q.Encode()

	req, err := http.NewRequest("POST", endpoint.String(), nil)
	if err != nil {
		panic(err)
	}
	req.Header.Set("Authorization", "Bearer "+os.Getenv("TXTFETCH_KEY"))

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	var result extractResponse
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		panic(err)
	}
	fmt.Println(result.ExtractedText)
}

{
  "status": "success",
  "extracted_text": "..."
}

large-or-slow-documents

POST /v1/extract returns 202 with a job_id whenever the document is large or slow to process, or whenever async=true is set. Poll GET /v1/extract/{job_id} for the result, or set webhook_url and have txtfetch push it instead. No single request on this page is safe to assume will always finish synchronously.

curl -X POST "https://api.txtfetch.com/v1/extract?url=https://example.com/report.pdf&async=true" \
  -H "Authorization: Bearer $TXTFETCH_KEY"

# {"status": "processing", "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6"}

curl "https://api.txtfetch.com/v1/extract/3fa85f64-5717-4562-b3fc-2c963f66afa6" \
  -H "Authorization: Bearer $TXTFETCH_KEY"

# {"status": "processing", "job_id": "..."} while running, then the same
# {"status": "success", "extracted_text": "...", "metadata": {...}} shape
# POST /v1/extract would have returned synchronously.

curl -X POST "https://api.txtfetch.com/v1/extract?url=https://example.com/report.pdf" \
  -H "Authorization: Bearer $TXTFETCH_KEY" \
  --data-urlencode "webhook_url=https://example.com/webhooks/txtfetch"

setup-steps

1. Add a task that calls the API, using SimpleHttpOperator or a PythonOperator running requests.post.
2. Set the endpoint to https://api.txtfetch.com/v1/extract with url as a query parameter set to each document's link.
3. Set the Authorization header to Bearer, followed by your API key, read from an Airflow connection or a secrets backend.
4. For a large batch, pass async set to true and either poll GET /v1/extract/{job_id} in a following task, or supply webhook_url instead.
5. Read extracted_text from the response and hand it to the next task with XCom.

faq

Is there an Airflow provider package for txtfetch?

No. txtfetch publishes no Airflow provider or operator. A DAG task built with SimpleHttpOperator or a plain requests.post call reaches the same REST API directly.

Should a task download each file before calling txtfetch?

No, not if the file already sits in reachable object storage. Pass its URL as the url query parameter and let txtfetch fetch it server-side, so the task never loads the full document into memory.

How does a DAG handle a batch with a mix of small and large documents?

Set async=true for the whole batch. Small documents still return a result quickly, and large ones return a job_id that a downstream task polls or a webhook_url delivers to.

Should a failed extraction fail the whole DAG run?

That depends on the pipeline. The task can check status and raise on a real error.code, while a retryable failure can lean on Airflow's own task retry settings instead.

related-reading

• Batch and large-document ingestion →
• Extract text from a PDF for RAG →
• Async jobs & webhooks →
• Idempotency →
• Get an API key →