Django

This guide covers a function-based synchronous view that uploads a file to /convert/raw, maps contract status codes from API to idiomatic Django responses, and a async def variant with httpx for Django 4.1+.

Synchronous view (function-based)

Use requests because Django’s synchronous views already run on a dedicated WSGI worker; there is no benefit in putting async in there.

# views.py
import time
import requests
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.views.decorators.http import require_POST
from django.views.decorators.csrf import csrf_exempt

API_URL = settings.FLASH_MD_API_URL  # "https://api.markpdf.tech"
API_KEY = settings.FLASH_MD_API_KEY

# Maximum size that we accept reading into memory before requiring BYOS
MAX_UPLOAD_BYTES = 25 * 1024 * 1024  # 25 MB


class ConversionError(Exception):
    """Conversion error mapped from an API status code."""

    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail
        super().__init__(detail)


def poll_job(job_id: str, timeout_seconds: int = 300) -> str:
    """Pollea GET /jobs/{job_id} cada ~5s hasta completed o failed."""
    deadline = time.monotonic() + timeout_seconds
    while True:
        if time.monotonic() > deadline:
            raise ConversionError(504, f"Timeout esperando el job {job_id}")

        time.sleep(5)
        res = requests.get(
            f"{API_URL}/jobs/{job_id}",
            headers={"x-api-key": API_KEY},
            timeout=30,
        )
        if res.status_code != 200:
            raise ConversionError(res.status_code, res.text)

        data = res.json()
        if data["status"] == "completed":
            return data["body"]
        if data["status"] == "failed":
            raise ConversionError(500, data.get("error", "El job falló"))
        # status == "processing" o "queued": seguimos esperando


def _map_upstream_error(status_code: int, body: str) -> JsonResponse:
    """Map a contract status code to a consistent Django response."""
    try:
        detail = requests.compat.json.loads(body).get("detail", body)
    except Exception:
        detail = body

    # We pass the same status code that returned the API except in special cases
    payload = {"error": detail}
    return JsonResponse(payload, status=status_code)


@csrf_exempt
@require_POST
def convert_document(request):
    uploaded = request.FILES.get("file")
    if not uploaded:
        return JsonResponse({"error": "Missing file ('file' field)"}, status=400)

    if uploaded.size > MAX_UPLOAD_BYTES:
        return JsonResponse(
            {"error": "File too large para este endpoint; usa BYOS (output_url)"},
            status=413,
        )

    try:
        res = requests.post(
            f"{API_URL}/convert/raw",
            params={"filename": uploaded.name, "mode": "fast"},
            headers={
                "x-api-key": API_KEY,
                "content-type": uploaded.content_type or "application/octet-stream",
            },
            data=uploaded.read(),
            timeout=300,
        )
    except requests.exceptions.Timeout:
        return JsonResponse({"error": "Timeout calling the conversion API"}, status=504)
    except requests.exceptions.ConnectionError:
        return JsonResponse({"error": "Could not connect to the conversion API"}, status=502)

    if res.status_code == 200:
        return HttpResponse(res.text, content_type="text/markdown; charset=utf-8")

    if res.status_code == 202:
        job = res.json()
        try:
            markdown = poll_job(job["job_id"])
        except ConversionError as exc:
            return JsonResponse({"error": exc.detail}, status=exc.status_code)
        return HttpResponse(markdown, content_type="text/markdown; charset=utf-8")

    # 400, 401, 403, 413, 415, 422, 429, 500: we pass the detail as is
    return _map_upstream_error(res.status_code, res.text)

The @csrf_exempt decorator assumes that this endpoint is consumed by an API client (mobile app or another service), not by a browser form within the same Django session. If you expose it to a form with a user session, use the normal CSRF token instead of exempting it.

Vista basada en clase (alternativa)

If you prefer the CBV style, the same flow fits well in a View:

# views.py
from django.views import View
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt


@method_decorator(csrf_exempt, name="dispatch")
class ConvertDocumentView(View):
    def post(self, request):
        uploaded = request.FILES.get("file")
        if not uploaded:
            return JsonResponse({"error": "Missing file ('file' field)"}, status=400)

        res = requests.post(
            f"{API_URL}/convert/raw",
            params={"filename": uploaded.name, "mode": "fast"},
            headers={
                "x-api-key": API_KEY,
                "content-type": uploaded.content_type or "application/octet-stream",
            },
            data=uploaded.read(),
            timeout=300,
        )

        if res.status_code == 200:
            return HttpResponse(res.text, content_type="text/markdown; charset=utf-8")
        if res.status_code == 202:
            markdown = poll_job(res.json()["job_id"])
            return HttpResponse(markdown, content_type="text/markdown; charset=utf-8")
        return _map_upstream_error(res.status_code, res.text)

Vista async (Django 4.1+)

Django 4.1+ supports async def in views. Combined with httpx.AsyncClient, avoid blocking the worker while waiting for a response from API (useful if your deployment uses ASGI, e.g. behind Uvicorn/Daphne).

# views.py
import asyncio
import httpx
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt

API_URL = settings.FLASH_MD_API_URL
API_KEY = settings.FLASH_MD_API_KEY


async def poll_job_async(client: httpx.AsyncClient, job_id: str) -> str:
    while True:
        await asyncio.sleep(5)
        res = await client.get(f"{API_URL}/jobs/{job_id}", headers={"x-api-key": API_KEY})
        data = res.json()
        if data["status"] == "completed":
            return data["body"]
        if data["status"] == "failed":
            raise RuntimeError(data.get("error", "El job falló"))


@csrf_exempt
async def convert_document_async(request):
    if request.method != "POST":
        return JsonResponse({"error": "Solo POST"}, status=405)

    uploaded = request.FILES.get("file")
    if not uploaded:
        return JsonResponse({"error": "Missing file ('file' field)"}, status=400)

    content = uploaded.read()

    async with httpx.AsyncClient(timeout=300) as client:
        res = await client.post(
            f"{API_URL}/convert/raw",
            params={"filename": uploaded.name, "mode": "fast"},
            headers={
                "x-api-key": API_KEY,
                "content-type": uploaded.content_type or "application/octet-stream",
            },
            content=content,
        )

        if res.status_code == 200:
            return HttpResponse(res.text, content_type="text/markdown; charset=utf-8")

        if res.status_code == 202:
            job = res.json()
            try:
                markdown = await poll_job_async(client, job["job_id"])
            except RuntimeError as exc:
                return JsonResponse({"error": str(exc)}, status=500)
            return HttpResponse(markdown, content_type="text/markdown; charset=utf-8")

        try:
            detail = res.json().get("detail", res.text)
        except Exception:
            detail = res.text
        return JsonResponse({"error": detail}, status=res.status_code)

request.FILES in Django requires that the body has been parsed by multipart/form-data. With ASGI and async views this still works the same (Django parses the form before invoking the view), but if your client uploads the file as a raw body without multipart, it reads request.body instead.

URLs

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path("api/convert/", views.convert_document),
    path("api/convert-async/", views.convert_document_async),
]

Settings

# settings.py
import os

FLASH_MD_API_URL = "https://api.markpdf.tech"
FLASH_MD_API_KEY = os.environ["FLASH_MD_API_KEY"]

Production

Timeouts

It will separate the connection timeout from the reading timeout. Large documents in quality mode may take longer than the default of requests:

import requests

res = requests.post(
    f"{API_URL}/convert/raw",
    params={"filename": "report.pdf", "mode": "balanced"},
    headers={"x-api-key": API_KEY, "content-type": "application/pdf"},
    data=file_bytes,
    timeout=(10, 300),  # (connect timeout, read timeout) en segundos
)

Retries with exponential backoff

Just retry 429 and 5xx. For the rest of the 4xx, I corrected the request instead of retrying:

import time
import random
import requests


def post_with_retry(url, max_retries=4, **kwargs):
    for attempt in range(max_retries + 1):
        try:
            res = requests.post(url, **kwargs)
        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
            if attempt == max_retries:
                raise
            time.sleep((2 ** attempt) + random.uniform(0, 1))
            continue

        if res.status_code == 429 or res.status_code >= 500:
            if attempt == max_retries:
                return res
            retry_after = res.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)
            continue

        return res

Large files with BYOS (`output_url`)

For large documents, avoid passing the entire Markdown through the body of your Django view: generate a pre-signed URL %0005%% from your own storage (S3, R2, etc.) and pass it as output_url. API uploads the result there directly and returns a small JSON instead of the full Markdown. This also auto-activates response_format=json, output_encoding=zstd and slim=true.

# views.py
import boto3
import requests
from django.conf import settings
from django.http import JsonResponse

s3 = boto3.client("s3", region_name="us-east-1")


def convert_large_document(request):
    uploaded = request.FILES.get("file")
    if not uploaded:
        return JsonResponse({"error": "Missing file"}, status=400)

    output_key = f"markdown/{uploaded.name}.md.zst"
    presigned_put_url = s3.generate_presigned_url(
        "put_object",
        Params={"Bucket": "mi-bucket", "Key": output_key},
        ExpiresIn=600,
    )
    presigned_head_url = s3.generate_presigned_url(
        "head_object",
        Params={"Bucket": "mi-bucket", "Key": output_key},
        ExpiresIn=600,
    )

    res = requests.post(
        f"{settings.FLASH_MD_API_URL}/convert/raw",
        params={
            "filename": uploaded.name,
            "mode": "fast",
            "output_url": presigned_put_url,
            "output_head_url": presigned_head_url,
        },
        headers={
            "x-api-key": settings.FLASH_MD_API_KEY,
            "content-type": uploaded.content_type or "application/octet-stream",
        },
        data=uploaded.read(),
        timeout=(10, 300),
    )

    if res.status_code not in (200, 202):
        return JsonResponse({"error": res.text}, status=res.status_code)

    # With output_url, the response 200/completed is a small JSON
    # which points to the object already uploaded to your bucket, not to the Markdown itself.
    return JsonResponse(res.json(), status=200)

If the same document is uploaded twice, output_head_url allows API to detect that the object already exists in your storage and return the result without reprocessing the document — useful for deduplicating repeated uploads without extra logic on your side.

Languages

Frameworks

Signed storage

Django

Django

Synchronous view (function-based)

Vista basada en clase (alternativa)

Vista async (Django 4.1+)

URLs

Settings

Production

Timeouts

Retries with exponential backoff

Large files with BYOS (`output_url`)

​Django

​Synchronous view (function-based)

​Vista basada en clase (alternativa)

​Vista async (Django 4.1+)

​URLs

​Settings

​Production

​Timeouts

​Retries with exponential backoff

​Large files with BYOS (output_url)

Django

Synchronous view (function-based)

Vista basada en clase (alternativa)

Vista async (Django 4.1+)

URLs

Settings

Production

Timeouts

Retries with exponential backoff

Large files with BYOS (`output_url`)