phase8: report generation isolated in worker (ARQ background job) + async endpoint + DMS output
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
This commit is contained in:
@@ -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)
|
||||
@@ -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."""
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user