from __future__ import annotations
from ppc_robot_lib import tasks
from ppc_robot_lib.output import ADAPTERS, OutputType, abstract_adapter
from ppc_robot_lib.output.types import OutputContext, SheetData, Table
from ppc_robot_lib.output.utils import build_data, build_header
from ppc_robot_lib.steps import AbstractStep
[docs]
class TablesOutputStep(AbstractStep):
"""
This step writes collection of tables as a spreadsheet (Excel), or writes them to external service (Google Sheets).
This step prepares all the data, loads previously stored state, and delegates the output process to a specified
adapter.
After the adapter completes (or fails with an exception), state is stored for further use for the current job.
"""
def __init__(self, tables: list[Table]):
"""
:param tables: List of tables to output. See :py:class:`ppc_robot_lib.utils.types.Table` for more details.
"""
self.tables = tables
def execute(self, task_ctx: tasks.TaskContextInterface) -> tasks.StepPerformance | None:
output_type = task_ctx.job.output_type
if not task_ctx.output_adapter:
adapter = self.create_adapter(output_type, task_ctx)
task_ctx.output_adapter = adapter
else:
adapter = task_ctx.output_adapter
rows_in, sheets = self.prepare_sheets(task_ctx)
adapter.write_output(task_ctx, sheets)
task_ctx.performance.add_output_rows(rows_in)
return tasks.StepPerformance(rows_in=rows_in)
def create_adapter(
self, output_type: OutputType, task_ctx: tasks.TaskContextInterface
) -> abstract_adapter.AbstractOutputAdapter:
"""
Loads previous output context and creates and initializes an output adapter of the given type.
:param output_type: Output type.
:param task_ctx: Task context.
:return:
"""
if output_type not in ADAPTERS:
available_adapters = ', '.join(adapter_type.value for adapter_type in ADAPTERS.keys())
raise ValueError(f'Adapter {output_type} is not recognized. Available adapters: {available_adapters}.')
job = task_ctx.job
output_path = job.output_path
previous_state = job.output_state_json
if not job.output_state_type or job.output_state_type == output_type:
output_ctx = OutputContext({}, output_path, previous_state)
else:
output_ctx = OutputContext({}, None, None)
adapter_type = ADAPTERS[output_type]
adapter = adapter_type(output_ctx)
adapter.initialize(task_ctx)
return adapter
def prepare_sheets(self, task_ctx: '') -> tuple[int, list[SheetData]]:
"""
Prepares :py:class:`SheetData` objects from the input tables.
:param task_ctx: Task Context.
:return: (total rows, list of sheet data)
"""
total_rows = 0
sheets = []
for table_def in self.tables:
if callable(table_def.sheet_name):
sheet_name = table_def.sheet_name(task_ctx)
else:
sheet_name = str(table_def.sheet_name)
table = task_ctx.work_set.get_table(table_def.table_name)
row_columns, header_rows, merges, formats = build_header(table_def.header, table)
header_row_count = len(header_rows)
data = build_data(row_columns, header_rows, table_def, table, task_ctx)
currency_column_index = None
if table_def.currency_column is not None:
for idx, row in enumerate(row_columns):
if row.name == table_def.currency_column:
currency_column_index = idx
break
sheets.append(
SheetData(
sheet_name,
row_columns,
merges,
formats,
data,
header_row_count,
table_def.sheet_id,
table_def.freeze_header,
table_def.freeze_columns,
table_def.row_height,
table_def.create_filter,
currency_column_index,
table_def.embedded_charts,
)
)
return total_rows, sheets