FastAPI Integration

pip install fastapi uvicorn document-stack

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from document_stack import AsyncDocumentStack

app = FastAPI()
ds_client = AsyncDocumentStack(api_key=os.environ["DS_API_KEY"])

class GenerateRequest(BaseModel):
    templateId: str
    data: dict = {}

class GenerateResponse(BaseModel):
    success: bool
    data: dict

@app.post("/api/generate", response_model=GenerateResponse)
async def generate_pdf(body: GenerateRequest):
    try:
        pdf = await ds_client.generate(
            template_id=body.templateId,
            data=body.data,
        )
        return GenerateResponse(
            success=True,
            data={"url": pdf.url},
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from fastapi.responses import StreamingResponse
import httpx

@app.get("/api/download/{template_id}")
async def download_pdf(template_id: str):
    pdf = await ds_client.generate(
        template_id=template_id,
        data={},
    )

    async with httpx.AsyncClient() as http:
        response = await http.get(pdf.url)

    return StreamingResponse(
        iter([response.content]),
        media_type="application/pdf",
        headers={
            "Content-Disposition": 'attachment; filename="document.pdf"',
        },
    )

import asyncio
from pydantic import BaseModel

class BatchItem(BaseModel):
    templateId: str
    data: dict = {}

class BatchRequest(BaseModel):
    items: list[BatchItem]

@app.post("/api/generate/batch")
async def batch_generate(body: BatchRequest):
    semaphore = asyncio.Semaphore(5)

    async def gen(item: BatchItem):
        async with semaphore:
            pdf = await ds_client.generate(
                template_id=item.templateId,
                data=item.data,
            )
            return {"templateId": item.templateId, "url": pdf.url}

    results = await asyncio.gather(
        *[gen(item) for item in body.items],
        return_exceptions=True,
    )

    successes = [r for r in results if isinstance(r, dict)]
    failures = [str(r) for r in results if isinstance(r, Exception)]

    return {
        "success": True,
        "data": {
            "generated": successes,
            "errors": failures,
        },
    }

from fastapi import Depends
from functools import lru_cache

@lru_cache
def get_ds_client():
    return AsyncDocumentStack(api_key=os.environ["DS_API_KEY"])

@app.post("/api/generate")
async def generate_pdf(
    body: GenerateRequest,
    client: AsyncDocumentStack = Depends(get_ds_client),
):
    pdf = await client.generate(
        template_id=body.templateId,
        data=body.data,
    )
    return {"success": True, "data": {"url": pdf.url}}