From a2c3f797f2b6ace553743d7487a371ec8d1775c5 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 29 Jul 2026 23:10:50 +0200 Subject: [PATCH] phase8: report generation isolated in worker (ARQ background job) + async endpoint + DMS output --- app/plugins/builtins/report_generator/jobs.py | 150 ++++++++++++++++++ .../builtins/report_generator/plugin.py | 3 + .../builtins/report_generator/routes.py | 44 +++++ 3 files changed, 197 insertions(+) create mode 100644 app/plugins/builtins/report_generator/jobs.py diff --git a/app/plugins/builtins/report_generator/jobs.py b/app/plugins/builtins/report_generator/jobs.py new file mode 100644 index 0000000..3af2820 --- /dev/null +++ b/app/plugins/builtins/report_generator/jobs.py @@ -0,0 +1,150 @@ +"""Background job for isolated report generation. + +This job runs in the ARQ worker process, NOT in the API process. +This isolates PDF/HTML rendering from the API: +- No shell access in worker +- No Docker socket +- No unnecessary secrets +- CPU/RAM limited by worker process +- Only approved templates +- URL fetching disabled (safe_url_fetcher) + +The job: +1. Fetches the template from DB +2. Renders the template (SandboxedEnvironment) +3. Generates PDF/CSV/Excel/JSON +4. Stores the result in DMS +5. Returns the DMS file ID +""" + +from __future__ import annotations + +import io +import logging +import uuid +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.db import create_db_session +from app.core.job_registry import register_job +from app.core.storage import get_storage_backend +from app.plugins.builtins.report_generator.models import ReportTemplate +from app.plugins.builtins.report_generator.pdf_generator import ( + generate_pdf_from_template_content, + render_template_string, +) + +logger = logging.getLogger(__name__) + + +def _generate_csv(rendered: str) -> io.BytesIO: + import csv + buf = io.BytesIO() + writer = csv.writer(io.TextIOWrapper(buf, encoding="utf-8", newline="")) + for line in rendered.strip().split("\n"): + writer.writerow(line.split(",")) + buf.seek(0) + return buf + + +def _generate_excel(rendered: str) -> io.BytesIO: + import openpyxl + wb = openpyxl.Workbook() + ws = wb.active + for i, line in enumerate(rendered.strip().split("\n"), 1): + for j, cell in enumerate(line.split(","), 1): + ws.cell(row=i, column=j, value=cell) + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + return buf + + +async def generate_report_job( + ctx: dict[str, Any], + tenant_id: str, + user_id: str, + template_id: str, + data: dict[str, Any], + output_format: str | None = None, +) -> dict[str, Any]: + """Background job: Generate a report and store in DMS. + + Returns: + {"dms_file_id": ..., "filename": ..., "format": ..., "size": ...} + """ + import hashlib + from app.plugins.builtins.dms.models import File as DmsFile + + async with create_db_session() as db: + # 1. Fetch template + result = await db.execute( + select(ReportTemplate).where( + ReportTemplate.id == uuid.UUID(template_id), + ReportTemplate.tenant_id == uuid.UUID(tenant_id), + ReportTemplate.deleted_at.is_(None), + ) + ) + template = result.scalar_one_or_none() + if template is None: + raise ValueError(f"Template {template_id} not found") + + fmt = output_format or template.output_format + + # 2. Generate content + if fmt in ("pdf", "print"): + raw_bytes, ext = generate_pdf_from_template_content( + template.content, data, fmt, + ) + elif fmt == "csv": + rendered = render_template_string(template.content, data) + file_data = _generate_csv(rendered) + raw_bytes = file_data.getvalue() + ext = "csv" + elif fmt == "excel": + rendered = render_template_string(template.content, data) + file_data = _generate_excel(rendered) + raw_bytes = file_data.getvalue() + ext = "xlsx" + elif fmt == "json": + rendered = render_template_string(template.content, data) + import json + raw_bytes = json.dumps(rendered, indent=2).encode() + ext = "json" + else: + raise ValueError(f"Unsupported format: {fmt}") + + # 3. Store in DMS + content_hash = hashlib.sha256(raw_bytes).hexdigest() + storage_path = f"reports/{tenant_id}/{uuid.uuid4().hex}.{ext}" + storage = get_storage_backend() + await storage.save(storage_path, raw_bytes) + + dms_file = DmsFile( + tenant_id=uuid.UUID(tenant_id), + name=f"{template.name}.{ext}", + folder_id=None, + uploaded_by=uuid.UUID(user_id), + mime_type="application/pdf" if fmt in ("pdf", "print") else f"text/{ext}", + size_bytes=len(raw_bytes), + storage_path=storage_path, + content_hash=content_hash, + owner_id=uuid.UUID(user_id), + ) + db.add(dms_file) + await db.flush() + await db.commit() + await db.refresh(dms_file) + + return { + "dms_file_id": str(dms_file.id), + "filename": dms_file.name, + "format": fmt, + "size": len(raw_bytes), + } + + +# Register with the job registry +register_job("generate_report", generate_report_job) diff --git a/app/plugins/builtins/report_generator/plugin.py b/app/plugins/builtins/report_generator/plugin.py index 2384928..8e69ccf 100644 --- a/app/plugins/builtins/report_generator/plugin.py +++ b/app/plugins/builtins/report_generator/plugin.py @@ -5,6 +5,9 @@ from __future__ import annotations from app.plugins.base import BasePlugin from app.plugins.manifest import PluginManifest, PluginRouteDef, FrontendMenuItem, FrontendPageRoute +# Register background jobs (isolated report generation in worker) +from app.plugins.builtins.report_generator import jobs # noqa: F401 + class ReportGeneratorPlugin(BasePlugin): """Report Generator plugin for generating reports from Jinja2 templates.""" diff --git a/app/plugins/builtins/report_generator/routes.py b/app/plugins/builtins/report_generator/routes.py index cb4fff2..002070f 100644 --- a/app/plugins/builtins/report_generator/routes.py +++ b/app/plugins/builtins/report_generator/routes.py @@ -410,6 +410,50 @@ async def generate_report( ) +@router.post("/generate/async") +async def generate_report_async( + body: ReportGenerateRequest, + current_user: dict = Depends(require_permission("reports:generate")), +): + """Submit a report generation job to the background worker. + + Returns a job ID. The report is generated in the isolated worker process + and stored in DMS. Poll /api/v1/reports/{job_id} for status. + """ + from arq import create_pool + from arq.connections import RedisSettings + from app.config import get_settings + + settings = get_settings() + tenant_id = current_user["tenant_id"] + user_id = current_user["user_id"] + + redis = await create_pool(RedisSettings( + host=settings.redis_host, + port=settings.redis_port, + password=settings.redis_password or None, + )) + + job = await redis.enqueue_job( + "generate_report", + tenant_id=tenant_id, + user_id=user_id, + template_id=body.template_id, + data=body.data, + output_format=body.output_format, + ) + + await redis.close() + + if job is None: + raise HTTPException( + 503, + detail={"detail": "Failed to enqueue report job", "code": "enqueue_failed"}, + ) + + return {"job_id": job.job_id, "status": "queued"} + + @router.get("/{report_id}") async def get_report( report_id: str,