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
[docs]
def write_tables(tables: list[Table]) -> None:
"""
Writes collection of tables as a spreadsheet (Excel), or writes them to external service (Google Sheets).
It 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.
This function can be called multiple times in a single report. You can
:param tables: List of tables to output. See :py:class:`ppc_robot_lib.utils.types.Table` for more details.
"""
task_ctx = tasks.get_context()
output_type = task_ctx.job.output_type
if not task_ctx.output_adapter:
adapter = _create_adapter(output_type, task_ctx)
task_ctx.output_adapter = adapter
else:
adapter = task_ctx.output_adapter
rows_in, sheets = _prepare_sheets(tables, task_ctx)
adapter.write_output(task_ctx, sheets)
def _create_adapter(
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: Initialized output adapter.
"""
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(tables: list[Table], task_ctx: 'tasks.TaskContextInterface') -> 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 tables:
if callable(table_def.sheet_name):
sheet_name = table_def.sheet_name(task_ctx)
else:
sheet_name = str(table_def.sheet_name)
if isinstance(table_def.table_name, str):
table = task_ctx.work_set.get_table(table_def.table_name)
else:
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