Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_scale_app_exception():
df1 = _some_df1()
scale_stage = Scale(
"StandardScaler", exclude_columns=[], exclude_object_columns=False
)
with pytest.raises(PipelineApplicationError):
scale_stage(df1)
df2 = _some_df2()
res_df = scale_stage(df2)
assert "ph" in res_df.columns
assert "gt" in res_df.columns
# test transform exception
with pytest.raises(PipelineApplicationError):
scale_stage(df1)
def test_colbyframefunc_error():
df = _some_df()
cbf_stage = ColByFrameFunc('A==B', _are_a_c_equal)
with pytest.raises(PipelineApplicationError):
cbf_stage(df)
def test_scale_app_exception():
df1 = _some_df1()
scale_stage = Scale(
"StandardScaler", exclude_columns=[], exclude_object_columns=False
)
with pytest.raises(PipelineApplicationError):
scale_stage(df1)
df2 = _some_df2()
res_df = scale_stage(df2)
assert "ph" in res_df.columns
assert "gt" in res_df.columns
# test transform exception
with pytest.raises(PipelineApplicationError):
scale_stage(df1)
if cols_to_exclude:
excluded = df[cols_to_exclude]
apply_to = df[
[col for col in df.columns if col not in cols_to_exclude]
]
else:
apply_to = df
self._scaler = scaler_by_params(self.scaler, **self._kwargs)
try:
res = pd.DataFrame(
data=self._scaler.fit_transform(apply_to),
index=apply_to.index,
columns=apply_to.columns,
)
except Exception:
raise PipelineApplicationError(
"Exception raised when Scale applied to columns {}".format(
apply_to.columns
)
)
if cols_to_exclude:
res = pd.concat([res, excluded], axis=1)
res = res[self._col_order]
self.is_fitted = True
return res
def _transform(self, df, verbose):
inter_df = df
try:
new_col = self._func(df)
except Exception:
raise PipelineApplicationError(
"Exception raised applying function{} to dataframe.".format(
self._func_desc
)
)
if self._follow_column:
loc = df.columns.get_loc(self._follow_column) + 1
else:
loc = len(df.columns)
inter_df = out_of_place_col_insert(
df=inter_df, series=new_col, loc=loc, column_name=self._column
)
return inter_df
self._col_order = list(df.columns)
if cols_to_exclude:
excluded = df[cols_to_exclude]
apply_to = df[
[col for col in df.columns if col not in cols_to_exclude]
]
else:
apply_to = df
try:
res = pd.DataFrame(
data=self._scaler.transform(apply_to),
index=apply_to.index,
columns=apply_to.columns,
)
except Exception:
raise PipelineApplicationError(
"Exception raised when Scale applied to columns {}".format(
apply_to.columns
)
)
if cols_to_exclude:
res = pd.concat([res, excluded], axis=1)
res = res[self._col_order]
return res