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_attribute_stage():
"""Testing attribute pipeline stages."""
pipeline = pdp.ColDrop('name').Bin({'speed': [5]}, drop=True)
assert isinstance(pipeline, PdPipeline)
assert isinstance(pipeline[0], ColDrop)
assert isinstance(pipeline[1], Bin)
df = _some_df()
res_df = pipeline(df)
assert 'speed' in res_df.columns
assert 'name' not in res_df.columns
def __getitem__(self, index):
if isinstance(index, slice):
return PdPipeline(self._stages[index])
return self._stages[index]
def __add__(self, other):
if isinstance(other, PdPipeline):
return PdPipeline([self, *other._stages])
elif isinstance(other, PdPipelineStage):
return PdPipeline([self, other])
else:
return NotImplemented
def __add__(self, other):
if isinstance(other, PdPipeline):
return PdPipeline([*self._stages, *other._stages])
elif isinstance(other, PdPipelineStage):
return PdPipeline([*self._stages, other])
else:
return NotImplemented
def __init__(self, stages, transformer_getter=None, **kwargs):
self._stages = stages
self._trans_getter = transformer_getter
self.is_fitted = False
super_kwargs = {
'exraise': False,
'exmsg': PdPipeline._DEF_EXC_MSG,
'appmsg': PdPipeline._DEF_APP_MSG
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)
def __init__(self, stages, transformer_getter=None, **kwargs):
self._stages = stages
self._trans_getter = transformer_getter
self.is_fitted = False
super_kwargs = {
'exraise': False,
'exmsg': PdPipeline._DEF_EXC_MSG,
'appmsg': PdPipeline._DEF_APP_MSG
}
super_kwargs.update(**kwargs)
super().__init__(**super_kwargs)
Parameters
----------
*stages : pdpipe.PipelineStage objects
PdPipeline stages given as positional arguments.
Returns
-------
p : pdpipe.PdPipeline
The resulting pipeline.
Examples
--------
import pdpipe as pdp
make_pdpipeline(pdp.ColDrop('a'), pdp.Bin('speed'))
"""
return PdPipeline(stages=stages)