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)

exception plexus.reports.service.ReportGenerationCancelled

Bases: RuntimeError

Raised when a report worker observes a dashboard cancellation request.

plexus.reports.service.decode_programmatic_run_payload(payload_base64: str) → Dict[str, Any]

plexus.reports.service.encode_programmatic_run_payload(payload: Dict[str, Any]) → str

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

plexus.reports.service.run_block_cached(block_class: str, block_config: dict, account_id: str, client: PlexusDashboardClient, cache_key: str | None = None, ttl_hours: float = 24, fresh: bool = False, background: bool = False, child_budget: Dict[str, Any] | None = None) → Tuple[Any | None, str | None, bool]

Run a report block, returning cached results when available.

Creates a Report + ReportBlock record to persist the result. On subsequent calls with the same cache_key within ttl_hours, returns the stored output without re-running the block.

When background is True and the cache misses, the block is queued as a durable Task for dispatcher execution and this function returns immediately with ({"status": "dispatched", "cache_key": cache_key, "task_id": ...}, None, False). A later call with the same cache_key and background=False (the default) will wait for the queued Task to finish and return its results when available.

Returns (output_data, log_string, was_cached).

plexus.reports.service.run_programmatic_block_and_persist(*, cache_key: str, block_class: str, block_config: Dict[str, Any], account_id: str, client: PlexusDashboardClient, persist_required: bool = False) → Tuple[Any | None, str | None]

plexus.reports.service.run_programmatic_report_and_persist(*, report_name: str, block_definitions: List[Dict[str, Any]], account_id: str, client: PlexusDashboardClient, report_parameters: Dict[str, Any] | None = None, display_title: str | None = None, display_subtitle: str | None = None, display_description: str | None = None) → Tuple[str | None, str | None]

Run multiple report blocks and persist them as a single Report with ordered ReportBlocks.