Flask Integration¶

from flask import Flask, request, jsonify
from tempfile import NamedTemporaryFile
from crxml import CrystalXMLSource, collect
import os

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 500 * 1024 * 1024  # 500 MB


@app.post("/parse-report")
def parse_report():
    if "file" not in request.files:
        return jsonify({"error": "No file provided"}), 400

    file = request.files["file"]
    if not file.filename:
        return jsonify({"error": "Empty filename"}), 400

    ext = os.path.splitext(file.filename or "")[1].lower()
    if ext not in (".xml", ".rpt"):
        return jsonify({"error": "Unsupported file type"}), 422

    with NamedTemporaryFile(delete=False, suffix=".xml") as tmp:
        tmp.write(file.read())
        tmp_path = tmp.name

    try:
        rows = collect(CrystalXMLSource(tmp_path))
        return jsonify({"rows": len(rows), "data": rows[:100]})
    except Exception as e:
        return jsonify({"error": str(e)}), 500
    finally:
        os.unlink(tmp_path)

from flask import Response
import csv
import io


@app.post("/to-csv")
def to_csv():
    file = request.files["file"]

    with NamedTemporaryFile(delete=False, suffix=".xml") as tmp:
        tmp.write(file.read())
        tmp_path = tmp.name

    def generate():
        src = CrystalXMLSource(tmp_path)
        writer = None
        for row in src:
            if writer is None:
                output = io.StringIO()
                writer = csv.DictWriter(output, fieldnames=list(row.keys()))
                writer.writeheader()
                yield output.getvalue()
            output = io.StringIO()
            writer = csv.DictWriter(output, fieldnames=list(row.keys()))
            writer.writerow(row)
            yield output.getvalue()

    return Response(generate(), mimetype="text/csv", headers={
        "Content-Disposition": "attachment; filename=report.csv"
    })

from flask import send_file
from openpyxl import Workbook
from io import BytesIO


@app.post("/to-xlsx")
def to_xlsx():
    file = request.files["file"]

    with NamedTemporaryFile(delete=False, suffix=".xml") as tmp:
        tmp.write(file.read())
        tmp_path = tmp.name

    try:
        rows = collect(CrystalXMLSource(tmp_path))
    finally:
        os.unlink(tmp_path)

    wb = Workbook()
    ws = wb.active
    if rows:
        ws.append(list(rows[0].keys()))
        for row in rows:
            ws.append(list(row.values()))

    buf = BytesIO()
    wb.save(buf)
    buf.seek(0)

    return send_file(buf, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                     as_attachment=True, download_name="report.xlsx")

app.config["CRXML_FIELD_MAP"] = {
    "{Report.InvoiceNo}": "invoice",
    "{Report.Customer}": "customer",
    "{Report.Amount}": "amount",
}


@app.post("/parse-mapped")
def parse_mapped():
    file = request.files["file"]
    mapping = app.config["CRXML_FIELD_MAP"]

    with NamedTemporaryFile(delete=False, suffix=".xml") as tmp:
        tmp.write(file.read())
        tmp_path = tmp.name

    try:
        src = CrystalXMLSource(tmp_path)
        pipe = src | RenameFields(mapping)
        rows = collect(pipe)
        return jsonify({"rows": len(rows), "data": rows[:100]})
    finally:
        os.unlink(tmp_path)

from crxml import UnpicklableStageError


@app.errorhandler(UnpicklableStageError)
def handle_unpicklable(e):
    return jsonify({"error": "Stage not compatible with parallel mode"}), 400


@app.errorhandler(ValueError)
def handle_value_error(e):
    return jsonify({"error": f"Parse error: {e}"}), 422