chunk-engine (py-chunks) in FastAPI, Flask, Django, Litestar, aiohttp, and Celery.

py-chunks ships upload-aware helpers so you rarely handle bytes yourself: get_chunks_from_upload for framework upload objects, get_chunks_from_bytes(data, filename) for raw bytes, plus the matching stream_chunks_from_* variants. See Input Sources.

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}

FastAPI — streaming response

Forward each chunk as NDJSON while the document is still being parsed:

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
from py_chunks import stream_chunks_from_upload
import json

app = FastAPI()

@app.post("/chunk/stream/")
async def chunk_stream(file: UploadFile = File(...)):
    def generate():
        for chunk in stream_chunks_from_upload(file):
            yield json.dumps(chunk) + "\n"
    return StreamingResponse(generate(), media_type="application/x-ndjson")

FastAPI — Markdown conversion

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

app = FastAPI()

@app.post("/markdown/")
async def to_markdown(file: UploadFile = File(...)):
    data = await file.read()
    return {"markdown": get_markdown(data, filename=file.filename)}

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}

Django

from django.http import JsonResponse
from py_chunks import get_chunks_from_upload

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

Litestar

from litestar import Litestar, post
from litestar.datastructures import UploadFile
from litestar.enums import RequestEncodingType
from litestar.params import Body
from py_chunks import get_chunks_from_bytes

@post("/chunk")
async def chunk_document(
    data: UploadFile = Body(media_type=RequestEncodingType.MULTI_PART),
) -> dict:
    content = await data.read()
    return {"chunks": get_chunks_from_bytes(content, data.filename)}

app = Litestar([chunk_document])

aiohttp

from aiohttp import web
from py_chunks import get_chunks_from_bytes

async def chunk_document(request):
    reader = await request.multipart()
    field = await reader.next()               # the "document" part
    data = await field.read(decode=False)
    chunks = get_chunks_from_bytes(data, field.filename)
    return web.json_response({"chunks": chunks})

app = web.Application()
app.add_routes([web.post("/chunk", chunk_document)])

Celery — background job

Chunking is CPU work; run it off the request path in a worker.

import celery
from py_chunks import get_chunks

@celery.task
def process_document(file_path: str):
    chunks = get_chunks(file_path, mode="semantic")
    # embed + persist to your vector store here
    return len(chunks)

Async uploads

If an upload object's read() is a coroutine (some ASGI stacks), get_chunks_from_upload raises TypeError. Read the bytes yourself and use the bytes API: data = await file.read(); get_chunks_from_bytes(data, file.filename). See Error Handling.

Python