Module rollup_cascade.column

define CascadeColumn (base) and RollupColumn (derived) column classes

Classes

class CascadeColumn (*, column: str, when_not_empty: collections.abc.Sequence[str] = (), default_value: str = '', strict: bool = False, custom_trigger: collections.abc.Callable[[numpy.ndarray, numpy.ndarray, collections.abc.Iterable[str]], bool] | None = None)

Specify a column whose value should be cascaded to subsequent rows when one or more reference columns has a value.

KwArgs

column : str

the column whose value should be cascaded

when_not_empty : Sequence[str]

the columns that trigger the cascade when not empty. if this setting itself is empty, the cascade is triggered for ALL rows. Default is ().

default_value : str

the value to cascade when no value has been captured for the cascade column. Default is "".

strict : bool

if True, ALL when_not_empty columns must be populated to trigger a cascade. Default is False.

custom_trigger : Callable | None

a callable accepting the current row value set, the next row value set, and an iterable of all column names

Subclasses

• RollupColumn

Class variables

var column : str

var custom_trigger : collections.abc.Callable[[numpy.ndarray, numpy.ndarray, collections.abc.Iterable[str]], bool] | None

var default_value : str

var strict : bool

var when_not_empty : collections.abc.Sequence[str]

Instance variables

prop check_when_not_empty : collections.abc.Callable[[numpy.ndarray], bool]

Retuns the callable for testing whether or not a row meets the when_not_empty constraint.

Methods

def cascade(self, table: dict[str, list[str]]) ‑> dict[str, list[str]]

Cascade values in the table for self.column per defined custom_trigger or when_not_empty column status.

Args

table

dict of lists of strings representing table rows

Returns

dict[str, list[str]]

table after cascading self.column.

def cascade_it(self, row_a: numpy.ndarray, row_b: numpy.ndarray, table: dict[str, list[str]]) ‑> bool

Returns True if value from row_a should be cascaded to row_b.

If self.custom_trigger is defined, call it with the supplied parameters and return the result. Otherwise, return self.set_it(row_b, table).

Args

row_a : np.ndarray

the last row with a good value.

row_b : np.ndarray

the target of the cascade, if triggered

table

dict of lists of strings representing table rows

Returns

bool

True if the last good value should be cascaded to the next row

def column_value(self, row: numpy.ndarray, table: dict[str, list[str]]) ‑> str

Retrieves the value of self.column from the supplied row.

Args

row : np.ndarray

the row to check

table

dict of lists of strings representing table rows

Returns

str

the value of self.column in the supplied row.

def header_check(self, table: dict[str, list[str]]) ‑> bool

Check the columns in self (column + when_not_empty) against those in the header of a real table to determine if this operation is relevant given the columns available.

Args

table

dict of lists of strings representing table rows

Returns

bool

indicating whether this Cascade/RollupColumn is applicable given the columns available in the table instance.

def meets_when_not_empty(self, row: numpy.ndarray, table: dict[str, list[str]]) ‑> bool

Tests the when_not_empty constraint on this row of values.

Args

row : np.ndarray

the row to check

table

dict of lists of strings representing table rows

Returns

bool

True if when_not_empty isn't defined, strict==False and a value is found at the position of any when_not_empty column, or strict is True and a value is present at the positions of ALL when_not_empty columns.

def set_column(self, value: str, row: numpy.ndarray, table: dict[str, list[str]]) ‑> numpy.ndarray

Set the value of self.column in the supplied row to the supplied value.

Args

value : str

the new value for self.column

row : np.ndarray

the row to check

table

dict of lists of strings representing table rows

Returns

np.ndarray

the updated row

def set_it(self, row: numpy.ndarray, table: dict[str, list[str]]) ‑> bool

Returns True if column is empty but the when_not_empty check is True. Used for defaulting and cascading.

Args

row : np.ndarray

the row to check

table

dict of lists of strings representing table rows

Returns

bool

True if this row's value should be set to the default.

class RollupColumn (*, column: str, when_not_empty: collections.abc.Sequence[str] = (), default_value: str = '', strict: bool = False, custom_trigger: collections.abc.Callable[[numpy.ndarray, numpy.ndarray, collections.abc.Iterable[str]], bool] | None = None, pre_cascade: bool = True, post_cascade: bool = True, join_function: collections.abc.Callable[[tuple[str, str]], str] = <built-in method join of str object>)

Specify a column whose values should be rolled up into the first row in which one or more reference columns is not empty.

KwArgs

column : str

the column whose value should be rolled up

when_not_empty : Sequence[str]

the columns that trigger the rollup when not empty.

default_value : str

the value to use when no value is present in the rollup column. Default is "".

strict : bool

if True, ALL when_not_empty columns must be populated to trigger a rollup. Default is False.

pre_cascade : bool

if True, the rollup is performed before column values have been cascaded. Default is True.

post_cascade : bool

if True, the rollup is performed after column values have been cascaded. Default is True.

join_function : Callable[[tuple[str, str]], str]

method to use to join the values when a rollup is triggered. Defaults to " ".join.

Ancestors

• CascadeColumn

Class variables

var post_cascade : bool

var pre_cascade : bool

Methods

def join_function(iterable, /) ‑> collections.abc.Callable[[tuple[str, str]], str]

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: '.'.join(['ab', 'pq', 'rs']) -> 'ab.pq.rs'

def roll_it(self, row_a: numpy.ndarray, row_b: numpy.ndarray, table: dict[str, list[str]]) ‑> bool

Check two subsequent table rows to see if row_b should be rolled into row_a.

Triggers rolling behavior when: - row_a meets our when_not_empty constraint, and - row_b does NOT meet our when_not_empty constraint, and - row_a has a value in the target column, and - row_b has a value in the target column, ** OR ** - only the target column is populated in both this and next row

Args

row_a : np.ndarray

the first row to check

row_b : np.ndarray

the second row to check

table

dict of lists of strings representing table rows

Returns

bool

True if row_b should be rolled into row_a

def rollup(self, table: dict[str, list[str]]) ‑> dict[str, list[str]]

Roll up values in the table for self.column per defined custom_trigger or when_not_empty column status.

Args

table

dict of lists of strings representing table rows

Returns

dict[str, list[str]]

table after rolling up self.column.

Inherited members

• CascadeColumn:
  • cascade
  • cascade_it
  • check_when_not_empty
  • column_value
  • header_check
  • meets_when_not_empty
  • set_column
  • set_it