msticpy.init.pivot_core.pivot_register module

Pivot helper functions .

class msticpy.init.pivot_core.pivot_register.PivotRegistration(input_type: str, entity_map: Dict[str, str], func_df_param_name: str | None = None, func_out_column_name: str | None = None, func_df_col_param_name: str | None = None, func_new_name: str | None = None, src_module: str | None = None, src_class: str | None = None, src_func_name: str | None = None, can_iterate: bool = True, func_static_params: Dict[str, Any] | None = None, func_input_value_arg: str | None = None, src_config_path: str | None = None, src_config_entry: str | None = None, entity_container_name: str | None = None, return_raw_output: bool = False, create_shortcut: bool = False)

Bases: object

Pivot registration for function.

Notes

src_modulestr

The src_module to import

src_classstr, optional

class to import and instantiate that contains the function/method (not needed if the target function is a pure Python function)

src_func_name: Callable

The function to wrap.

func_new_name: str, optional

Rename the function to this, defaults to src_func_name

input_typestr

The input data type that the function is expecting. One of ‘dataframe’, ‘iterable’, ‘value’

can_iterate: bool, optional

True if the function supports being called multiple times (for iterable input). Default is True

entity_map: Dict[str, str]

dict of entities supported (keys) and attribute to use from entity as input to the function

func_df_param_name: str

The name of the parameter that func takes the input value e.g. func(ip=my_address) => ‘ip’ == func_df_col_param_name. In the case of a DataFrame, this is usually ‘data’

func_df_col_param_name: str

The name that the target function uses to identify the column to use for input in the input DataFrame.

func_out_column_name: str, optional

The name of the column in the output DF to use as a key to join to the input. If None, use func_df_col_param_name

func_static_params: Optional[Dict[str, Any]]

static parameters (kwargs) that are always passed to the target function

func_input_value_arg: Optional[str]

The name of kwarg passed to the function that contain the input value. If function supports DF input, func_df_col_param_name will be used and this is not needed.

src_config_pathOptional[str]

The source path that the configuration was read from, default None.

src_config_entryOptional[str]

The entry name in the configuration file, default None.

entity_container_nameOptional[str]

The name of the container in the entity that will hold this pivot function.

return_raw_outputbool

Return raw output from the wrapped function, do not try to format into a DataFrame. Default is False.

create_shortcutbool

If True, create a shortcut function directly on the entity.

Method generated by attrs for class PivotRegistration.

attr_for_entity(entity: Entity | str) → str | None

Return the attribute to use for the specified entity.

Parameters:

entity (Union[entities.Entity, str]) – Entity instance or name

Returns:

Attribute name to use.

Return type:

Optional[str]

can_iterate: bool

create_shortcut: bool

entity_container_name: str | None

entity_map: Dict[str, str]

func_df_col_param_name: str | None

func_df_param_name: str | None

func_input_value_arg: str | None

func_new_name: str | None

func_out_column_name: str | None

func_static_params: Dict[str, Any] | None

input_type: str

return_raw_output: bool

src_class: str | None

src_config_entry: str | None

src_config_path: str | None

src_func_name: str | None

src_module: str | None

msticpy.init.pivot_core.pivot_register.create_pivot_func(target_func: Callable[[Any], Any], pivot_reg: PivotRegistration) → Callable[[...], DataFrame]

Create function wrapper for pivot function.

Parameters:

• target_func (Callable) – The target function to wrap.

• pivot_reg (PivotRegistration) – The pivot function registration object.

Returns:

The original target_func wrapped in pre-processing and post-processing code.

Return type:

Callable[[Any], pd.DataFrame]

msticpy.init.pivot_core.pivot_register.get_join_params(func_kwargs: Dict[str, Any]) → Tuple[str | None, str | None, str | None, bool]

Get join parameters from kwargs.

Parameters:

func_kwargs (Dict[str, Any]) – Keyword arguments from caller

Returns:

join_type, left_on, right_on, join_ignore_case

Return type:

Tuple[str, str, str, bool]

msticpy.init.pivot_core.pivot_register.join_result(input_df: DataFrame, result_df: DataFrame, how: str, left_on: str, right_on: str, ignore_case: bool) → DataFrame

Join input and result DFs, optionally ignoring case.

Parameters:

• input_df (pd.DataFrame) – Input DF

• result_df (pd.DataFrame) – Result DF

• how (str) – Join type - “inner”, “left”, “right”, “outer”

• left_on (str) – Column from input_df to use as join key

• right_on (str) – Column from result_df to use as join key

• ignore_case (bool) – If True and input_df column is a string

Returns:

The merged DataFrame

Return type:

pd.DataFrame