flexorch-sdk

Python SDK for the FlexOrch API.

FlexOrch turns unstructured documents (PDF, DOCX, invoices, emails…) into clean, structured, LLM-ready datasets — with automatic PII detection and masking, quality scoring, and multiple export formats.

---

Install

pip install flexorch-sdk

Requires Python 3.10+. The only dependency is httpx.

---

Quick start

from flexorch_sdk import FlexOrchClient

client = FlexOrchClient("fx_your_key_here")

# Upload a document and wait for the pipeline to finish
job = client.process("contract.pdf", locale="tr").wait()

print(job.quality_grade)   # "A"
print(job.quality_score)   # 0.91

# Build a dataset from the job, then download it
dataset = job.build_dataset().wait().dataset()
dataset.export("jsonl", path="output.jsonl")

---

Auth

Pass your API key directly or set the FLEXORCH_API_KEY environment variable:

export FLEXORCH_API_KEY=fx_...

from flexorch_sdk import FlexOrchClient

client = FlexOrchClient()   # reads FLEXORCH_API_KEY automatically

Get your API key from app.flexorch.com → Settings.

---

Supported input formats

Category	Formats
Documents	PDF (text + scanned), DOCX, TXT
Spreadsheets	XLSX
Email	EML, MSG
E-invoices	XML/UBL (Peppol, GİB TR), FatturaPA (IT), XRechnung (DE), ZUGFeRD/Factur-X
Images	JPG, PNG, TIFF (OCR)
Web	HTML, HTM

---

Export formats

json · jsonl · csv · parquet · md · xml · xlsx · rag · hf

dataset.export("jsonl", path="output.jsonl")   # write to file
raw = dataset.export("parquet")                # return bytes
dataset.export("rag", min_quality="B")         # only A/B-grade chunks

The rag format produces LlamaIndex/LangChain-compatible chunks with metadata. The hf format is a zip archive readable with datasets.load_from_disk().

---

Processing

Single file

job = client.process("invoice.pdf", locale="de").wait()

locale is an IETF language tag used to activate the right PII detectors (tr, de, en, fr, it, nl, es, pl, und = all).

Batch

jobs = client.process_many(["a.pdf", "b.pdf", "c.pdf"], locale="und")
for job in jobs:
    job.wait()
    print(job.quality_grade, job.quality_score)

From S3

# Register a connector once; store conn.id for reuse
conn = client.connectors.create(
    "Production S3", "s3",
    {
        "bucket": "my-bucket",
        "region": "eu-central-1",
        "access_key_id": "AKIA...",
        "secret_access_key": "...",
    },
)

# Verify connectivity
result = client.connectors.test(conn.id)
print(result.success, result.latency_ms)   # True, 38

# Process files from S3
jobs = client.process_from_s3(conn.id, ["invoices/inv-001.pdf", "invoices/inv-002.pdf"])
for job in jobs:
    job.wait()

---

Job polling

Job.wait() blocks until the pipeline completes or times out.

job = client.process("large-report.pdf").wait(
    timeout=600,       # seconds before TimeoutError (default: 300)
    poll_interval=5,   # polling interval in seconds (default: 2)
)

print(job.status)        # "completed"
print(job.quality_grade) # "A" | "B" | "C" | "D"
print(job.quality_score) # 0.0 – 1.0
print(job.has_dataset)   # False — no dataset exists yet, see below
print(job.execution_id)  # needed by build_dataset() / build_from_execution()
print(job.degraded)      # False — True if structured extraction failed but
                          # the job still completed (PII/quality results are
                          # still meaningful; job.wait() does not raise for this)

---

Building a dataset

A completed job does not have a dataset until you build one — this is a separate, explicit step (it's what lets you build one dataset from several jobs, or re-run it with force_rebuild=True):

job = client.process("contract.pdf").wait()

build_job = job.build_dataset(name="contracts-q1")  # or client.datasets.build_from_execution(job.execution_id, ...)
ds = build_job.wait().dataset()

Dataset operations

ds = client.datasets.get("dataset-id")

print(ds.name)              # "contract-2024-q1"
print(ds.row_count)         # 142
print(ds.available_formats) # ["json", "jsonl", "csv", "parquet"]

# Download locally
ds.export("jsonl", path="output.jsonl")

# Push directly to S3
push = ds.export_to_s3(conn.id, "jsonl", prefix="processed/datasets/")
print(push["s3_key"])       # "processed/datasets/contract-2024-q1.jsonl"
print(push["size_bytes"])   # 84320

# Semantic indexing (Pro+)
ds.index()
status = ds.index_status()  # {"status": "ready", "chunks_indexed": 48}

# Preview rows, quality/privacy profile, KVKK/GDPR compliance report
rows = ds.rows(page=1, page_size=50)
profile = ds.profile()
report = ds.compliance_report()   # Pro+ required

---

Semantic search (Pro+)

results = client.search(
    "payment terms net 30",
    top_k=10,
    filters={
        "document_type": "invoice",
        "language": "de",
        "quality_grade": "A",
        "pii_masked": True,
    },
)

for r in results:
    print(f"{r.score:.3f}  [{r.dataset_id}]  {r.text[:120]}")

---

Resources

# Jobs
jobs = client.jobs.list(page=1, page_size=20)
job  = client.jobs.get("job-id")
client.jobs.submit_feedback("job-id", "down", issue="missing_fields", notes="PO number not extracted")
feedback = client.jobs.get_feedback("job-id")  # None if not submitted yet

# Documents
docs = client.documents.list(page=1, page_size=20)
doc  = client.documents.get("document-id")     # includes processing_history, related_datasets
reprocess_job = doc.reprocess()

# Datasets
datasets = client.datasets.list()
ds       = client.datasets.get("dataset-id")

# Usage
usage = client.usage.current()
print(f"{usage.credits_used} / {usage.credits_limit} credits used  (plan: {usage.plan})")
if usage.is_trial:
    print(f"{usage.trial_days_remaining} trial days left")

history = client.usage.history(period="30d")        # daily credits + job counts
trend   = client.usage.quality_trend(period="30d")   # daily avg quality score
limits  = client.usage.rate_limits()                 # current window usage, doesn't consume a slot

# Webhooks
client.webhooks.register("https://your-server.com/hook", events=["dataset.ready"])
client.webhooks.list()
client.webhooks.delete("webhook-id")

# Connectors
client.connectors.create("name", "s3", {...})
client.connectors.list()
client.connectors.get("connector-id")
client.connectors.test("connector-id")
client.connectors.delete("connector-id")

# Connector schedules (Pro+)
schedule = client.connectors.create_schedule("connector-id", "0 2 * * *", prefix_filter="invoices/")
client.connectors.list_schedules("connector-id")
client.connectors.trigger_schedule("connector-id", schedule.id)   # run now instead of waiting for cron
client.connectors.schedule_logs("connector-id", schedule.id)
client.connectors.delete_schedule("connector-id", schedule.id)

---

Error handling

from flexorch_sdk import (
    FlexOrchClient,
    AuthError,       # 401 — invalid or missing API key
    QuotaError,      # 402 — credit limit reached or trial expired
    RateLimitError,  # 429 — too many requests; has .retry_after (seconds)
    NotFoundError,   # 404
    ValidationError, # 422 — bad request parameters
    ServerError,     # 5xx
    JobFailedError,  # pipeline failed; has .job_id and .failure_reason
    TimeoutError,    # Job.wait() exceeded timeout; has .job_id
)

try:
    job = client.process("doc.pdf").wait(timeout=120)
except AuthError:
    print("Invalid API key — check FLEXORCH_API_KEY")
except QuotaError as e:
    print(f"Out of credits — reset at {e.reset_at}")
except JobFailedError as e:
    print(f"Pipeline failed for job {e.job_id}: {e.failure_reason}")
except TimeoutError as e:
    print(f"Job {e.job_id} still running after timeout — poll manually")

The SDK automatically retries 429 and 5xx responses with exponential backoff (up to 3 attempts by default).

---

Configuration

client = FlexOrchClient(
    api_key="fx_...",
    base_url="https://api.flexorch.com/v1",  # override for self-hosted
    timeout=60.0,       # HTTP timeout per request in seconds
    max_retries=5,      # retry attempts for transient errors
)

Context manager

with FlexOrchClient() as client:
    job = client.process("report.pdf").wait()
    job.dataset().export("jsonl", path="report.jsonl")
# HTTP connection pool released automatically

---

Examples

See examples/ for runnable scripts:

File	Description
basic_process.py	Process a single document and export as JSONL
batch_process.py	Process multiple files with error handling
s3_import.py	Import from S3, process, export results back to S3

---

Development

git clone https://github.com/flexorch/flexorch-sdk
cd flexorch-sdk
pip install -e ".[dev]"
pytest

Tests use respx to mock httpx — no network calls, no API key needed.

---

Links

• Platform
• API reference
• flexorch-audit — open-source PII detection library

---

License

MIT