py-chunks 0.1.0

Rust-backed Python chunking library

py-chunks

Fast, framework-agnostic document chunking library backed by Rust. Extract meaningful content segments from DOCX, PDF, PPTX, TXT, MD, and HTML files—optimized for production use.

Features

• 6 Document Formats: DOCX, PDF, PPTX, TXT, Markdown, HTML
• Zero Dependencies: Pure Python stdlib; all parsing is Rust-compiled
• Framework Agnostic: Local files, bytes, file-like objects, FastAPI uploads, S3 presigned URLs, or raw streams
• Consistent Output Schema: Structured chunks with content type, section headings, and format-specific metadata
• High Performance: Rust parsing engine with millisecond latency per document
• Production Ready: Full type hints, comprehensive error handling, 15 integration tests, Pylint 10.00/10

Installation

pip install py-chunks

Requirements: Python 3.9+

Quick Start

from py_chunks import get_chunks

# Auto-detects source type and format
chunks = get_chunks("document.pdf")

for chunk in chunks:
    print(chunk["content"])
    print(chunk["content_type"])  # "heading", "plain_paragraph", etc.

Supported Sources

The library accepts documents from multiple sources with automatic type detection:

Source	Method	Example
Local File Path	get_chunks() or get_chunks_from_path()	get_chunks("report.pdf")
Raw Bytes	get_chunks_from_bytes()	get_chunks_from_bytes(data, filename="report.pdf")
File-like Object	get_chunks_from_fileobj()	get_chunks_from_fileobj(BytesIO(...), filename="doc.md")
FastAPI Upload	get_chunks_from_upload()	get_chunks_from_upload(file)
S3 Presigned URL	get_chunks_from_s3_presigned_url()	get_chunks_from_s3_presigned_url(url)

Supported Formats

Format	Extensions
PDF	.pdf
DOCX	.docx
PPTX	.pptx
Markdown	.md
HTML	.html, .htm
Plain Text	.txt

Usage Examples

From local file:

from py_chunks import get_chunks

chunks = get_chunks("report.pdf")

From bytes (API upload):

from py_chunks import get_chunks_from_bytes

file_bytes = request.files['document'].read()
chunks = get_chunks_from_bytes(file_bytes, filename="report.pdf")

From FastAPI upload:

from fastapi import FastAPI, File, UploadFile
from py_chunks import get_chunks_from_upload

app = FastAPI()

@app.post("/upload/")
async def chunk_document(file: UploadFile = File(...)):
    chunks = get_chunks_from_upload(file)
    return {"chunks": chunks}

From file-like object:

from py_chunks import get_chunks_from_fileobj
from io import BytesIO

bio = BytesIO(file_data)
chunks = get_chunks_from_fileobj(bio, filename="document.md")

From S3 presigned URL:

from py_chunks import get_chunks_from_s3_presigned_url

url = "https://bucket.s3.amazonaws.com/file.docx?AWSAccessKeyId=..."
chunks = get_chunks_from_s3_presigned_url(url)

API Reference

Universal Entry Point

get_chunks(source, *, filename: str | None = None) -> list[dict]

Accepts any source type and automatically detects format and dispatch target.

Parameters:

• source: str, PathLike, bytes, file-like object, upload object, or URL
• filename: Optional filename to disambiguate format (required for bytes/file objects without .filename attribute)

Returns: list[dict] – Each chunk is a dictionary with:

• content: str – Extracted text segment
• content_type: str – Content classification
• metadata: dict – Section heading, page number, source type, format-specific fields

Raises: FileNotFoundError, ValueError, TypeError with descriptive messages

Format-Specific Entry Points

• get_chunks_from_path(file_path: str) -> list[dict] – Local filesystem
• get_chunks_from_bytes(data: bytes, filename: str) -> list[dict] – Raw bytes (temp file auto-cleanup)
• get_chunks_from_fileobj(file_obj, filename: str | None = None) -> list[dict] – File-like objects (open(), BytesIO, etc.)
• get_chunks_from_upload(upload_file) -> list[dict] – FastAPI/Starlette uploads (handles async)
• get_chunks_from_s3_presigned_url(url: str, filename: str | None = None, timeout: int = 60) -> list[dict] – HTTP/HTTPS URLs

Advanced: Format-Specific Chunkers

Direct access to format parsers (returns timing information):

from py_chunks.chunkers import chunk_pdf, chunk_docx, chunk_html, chunk_md, chunk_pptx, chunk_txt

chunks, timing = chunk_pdf("file.pdf")
print(f"Rust: {timing['rust_ms']}ms, Python: {timing['python_ms']}ms")

Each format-specific chunker returns a tuple: (list[dict], dict) where the second element contains rust_ms and python_ms keys.

Output Schema

Chunk Structure

{
    "content": "The extracted text segment.",
    "content_type": "plain_paragraph",  # See content types below
    "metadata": {
        "section_heading": "Section Title" | None,
        "footnotes_captions": [...],
        "page_number": 1 | None,
        "document_metadata": {
            "source_type": "pdf",
            # format-specific fields
        }
    }
}

Content Types

• heading – Section heading (H1-H6, bold text, etc.)
• plain_paragraph – Prose paragraph
• bullet_list – Unordered list
• table – Tabular data
• code_block – Code or preformatted text
• long_single_paragraph – Paragraph > 900 characters
• short_disconnected_paragraph – Paragraph < 90 characters

Architecture

System Design

┌──────────────────────────────────┐
│     Python Public API            │
│   (py_chunks/__init__.py)        │
│  ├─ get_chunks()                 │
│  ├─ get_chunks_from_path()       │
│  ├─ get_chunks_from_bytes()      │
│  ├─ get_chunks_from_fileobj()    │
│  ├─ get_chunks_from_upload()     │
│  └─ get_chunks_from_s3_presigned_url() │
└────────────┬─────────────────────┘
             │
             ↓
┌──────────────────────────────────┐
│    Format Dispatcher             │
│  (py_chunks/chunkers/*.py)       │
│  ├─ chunk_pdf()                  │
│  ├─ chunk_docx()                 │
│  ├─ chunk_pptx()                 │
│  ├─ chunk_html()                 │
│  ├─ chunk_md()                   │
│  └─ chunk_txt()                  │
└────────────┬─────────────────────┘
             │
             ↓
┌──────────────────────────────────┐
│     Rust Extension Module        │
│    (src/extensions/*.rs)         │
│  • Format-specific parsing       │
│  • Returns PyDict chunks         │
│  • Timing instrumentation        │
└──────────────────────────────────┘

Design Principles

• Single Responsibility: Each format parser handles only its format
• Framework Agnostic: Python layer abstracts source complexity; Rust layer is format-specific
• Temp File Strategy: Bytes → temp file → Rust → auto-cleanup provides safety without complex FFI
• Minimal Dependencies: Requests for S3 URLs only; all parsing is Rust-compiled

Development & Testing

Running Tests

cd py_chunks
python -m pytest -v

Expected: 15 integration tests passing

Code Quality

python -m pylint py_chunks tests/test_source_apis.py

Expected: 10.00/10 score

Local Development Build

pip install -e .

Rebuilds the Rust extension with local changes.

Error Handling

All functions raise descriptive exceptions:

from py_chunks import get_chunks

try:
    chunks = get_chunks("missing.pdf")
except FileNotFoundError as e:
    print(e)  # File not found: missing.pdf

try:
    chunks = get_chunks("image.png")
except ValueError as e:
    print(e)  # Unsupported file type '.png'. Supported: .docx, .htm, .html, .md, .pdf, .pptx, .txt

Framework Integration

Flask

from flask import Flask, request
from py_chunks import get_chunks_from_bytes

app = Flask(__name__)

@app.post("/chunk")
def chunk_document():
    file = request.files['document']
    chunks = get_chunks_from_bytes(file.read(), file.filename)
    return {"chunks": chunks}

FastAPI

from fastapi import FastAPI, File, UploadFile
from py_chunks import get_chunks_from_upload

app = FastAPI()

@app.post("/chunk/")
async def chunk_document(file: UploadFile = File(...)):
    chunks = get_chunks_from_upload(file)
    return {"chunks": chunks}

Django

from django.http import JsonResponse
from py_chunks import get_chunks_from_upload

def chunk_view(request):
    if request.FILES:
        file = request.FILES['document']
        chunks = get_chunks_from_upload(file)
        return JsonResponse({"chunks": chunks})
    return JsonResponse({"error": "No file"}, status=400)

Celery Background Job

import celery
from py_chunks import get_chunks

@celery.task
def process_document(file_path: str):
    chunks = get_chunks(file_path)
    # Persist to database
    return len(chunks)

License

MIT

---

Built with Rust (performance) + Python (simplicity)