from ppc_robot_lib.steps.abstract_step import AbstractStep
from ppc_robot_lib.tasks.task_context import TaskContextInterface, StepPerformance
from ppc_robot_lib.utils.types import Order, OrderBy
[docs]
class SortStep(AbstractStep):
"""
Sorts the table by one or more columns.
**Example:**
>>> from ppc_robot_lib.steps.transformations import SortStep
>>> from ppc_robot_lib.utils.types import Order, OrderBy
>>> SortStep("in_table", order=[
... OrderBy("Impressions", Order.DESC),
... OrderBy("CampaignName", Order.ASC),
... ])
"""
def __init__(self, input_table: str, order: list[OrderBy] | tuple[OrderBy, ...], output_table: str = None):
"""
:param input_table: Input table.
:param order: List of :py:class:`ppc_robot_lib.utils.types.OrderBy` objects that specify the sorting order.
:param output_table: Output table. If none is given, input table is used.
"""
if len(order) == 0:
raise ValueError('Empty set of sort columns is not allowed.')
self.input_table = input_table
self.order = order
if output_table is not None:
self.output_table = output_table
else:
self.output_table = input_table
def execute(self, task_ctx: TaskContextInterface) -> StepPerformance:
inplace = self.input_table == self.output_table
table = task_ctx.work_set.get_table(self.input_table)
columns = []
ascending = []
for order_by in self.order:
columns.append(order_by.column)
ascending.append(False if order_by.order is Order.DESC else True)
result = table.sort_values(by=columns, ascending=ascending, inplace=inplace)
if inplace:
new_table = table
else:
new_table = result
if self.output_table in task_ctx.work_set:
task_ctx.work_set.delete_table(self.output_table)
task_ctx.work_set.set_table(self.output_table, new_table)
rows = len(new_table.index)
return StepPerformance(new_table, rows_in=rows, rows_out=rows)