plexus.reports.service module

class plexus.reports.service.ReportBlockExtractor

Bases: BaseRenderer

__init__()

block_code(token, state)

finalize(data)

heading(token, state)

name = 'report_block_extractor'

paragraph(token, state)

render_tokens(tokens, state)

text(token, state)

plexus.reports.service.generate_report(task_id: str)

Celery task handler for generating a report based on a Task record. Fetches task, initializes tracker, calls core logic, and sets final task status.

Args:

task_id: The ID of the Task record triggering this report generation.

Returns:

None. Task status indicates success or failure.

plexus.reports.service.generate_report_with_parameters(config_id: str, parameters: Dict[str, Any], account_id: str, client: PlexusDashboardClient, trigger: str = 'api', task_object: Task | None = None, log_prefix: str | None = None) → Tuple[str, str | None, str]

Complete report generation workflow with parameter handling.

This is the shared implementation used by both CLI and MCP interfaces. It handles parameter validation, Jinja2 rendering, task creation, and report generation.

Args:

config_id: Report configuration ID parameters: Dictionary of parameter values (pre-collected, not interactive) account_id: Account ID owning the report client: Dashboard API client trigger: Source of the generation request (cli_sync, mcp, etc.) task_object: Optional existing Task object. If None, creates new task. log_prefix: Optional log prefix override

Returns:

Tuple of (report_id, error_message, task_id) - report_id: ID of created report - error_message: First block error message (None if success) - task_id: ID of the task tracking this generation

Raises:

ValueError: If configuration not found, parameters invalid, or rendering fails