Skip to content

Scrubber¤

The main class of the submodule, containing the main functions users will use.

Pipe ¤

A Pydantic dataclass containing a pipeline component.

Calls

The class is callable and returns a function that takes a string and returns a string.

Methods:

Name Description
__call__

Call the pipeline component on the text.
Source code in lexos/scrubber/scrubber.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class Pipe:
    """A Pydantic dataclass containing a pipeline component.

    Calls:
        The class is callable and returns a function that takes a string and returns a string.
    """

    name: str = Field(..., description="The name of the component.")
    opts: Optional[dict[str, Any]] = Field(
        default={}, description="Options to pass to the component."
    )
    factory: Optional[catalogue.Registry] = Field(
        default_factory=lambda: scrubber_components,
        description="The factory to use to get the component.",
    )

    def __call__(self, text: str) -> Callable:
        """Call the pipeline component on the text.

        Args:
            text (str): The text to scrub.

        Returns:
            Callable: A function that takes a string and returns a string.
        """
        try:
            func = self.factory.get(self.name)
            return func(text, **self.opts)
        except NameError as e:
            raise LexosException(e)
        except catalogue.RegistryError as e:
            raise LexosException(e)

__call__(text: str) -> Callable ¤

Call the pipeline component on the text.

Parameters:

Name Type Description Default
text str

The text to scrub.

 required

Returns:

Name Type Description
Callable Callable

A function that takes a string and returns a string.
Source code in lexos/scrubber/scrubber.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __call__(self, text: str) -> Callable:
    """Call the pipeline component on the text.

    Args:
        text (str): The text to scrub.

    Returns:
        Callable: A function that takes a string and returns a string.
    """
    try:
        func = self.factory.get(self.name)
        return func(text, **self.opts)
    except NameError as e:
        raise LexosException(e)
    except catalogue.RegistryError as e:
        raise LexosException(e)

name: str = Field(..., description='The name of the component.') class-attribute instance-attribute ¤

opts: Optional[dict[str, Any]] = Field(default={}, description='Options to pass to the component.') class-attribute instance-attribute ¤

factory: Optional[catalogue.Registry] = Field(default_factory=(lambda: scrubber_components), description='The factory to use to get the component.') class-attribute instance-attribute ¤

__call__(text: str) -> Callable ¤

Call the pipeline component on the text.

Parameters:

Name Type Description Default
text str

The text to scrub.

 required

Returns:

Name Type Description
Callable Callable

A function that takes a string and returns a string.
Source code in lexos/scrubber/scrubber.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __call__(self, text: str) -> Callable:
    """Call the pipeline component on the text.

    Args:
        text (str): The text to scrub.

    Returns:
        Callable: A function that takes a string and returns a string.
    """
    try:
        func = self.factory.get(self.name)
        return func(text, **self.opts)
    except NameError as e:
        raise LexosException(e)
    except catalogue.RegistryError as e:
        raise LexosException(e)

Scrubber ¤

A class to scrub text using a pipeline of components.

Methods:

Name Description
__init__

Initialize the Scrubber object.
add_pipe

Add a component to the scrubber pipeline.
pipe

Scrub a list of texts with the current pipeline.
remove_pipe

Remove a component from the scrubber pipeline.
reset

Remove all components from the pipeline.
scrub

Run a text through the scrubber pipeline.

Attributes:

Name Type Description
pipes list[Pipe]

Return a list of the names of the pipeline components.
Source code in lexos/scrubber/scrubber.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class Scrubber:
    """A class to scrub text using a pipeline of components."""

    def __init__(self) -> None:
        """Initialize the Scrubber object."""
        self._components: list[Pipe] = []

    @property
    def pipes(self) -> list[Pipe]:
        """Return a list of the names of the pipeline components."""
        return [component.name for component in self._components]

    def _get_pipe_index(
        self,
        before: Optional[str | int] = None,
        after: Optional[str | int] = None,
        first: Optional[bool] = None,
        last: Optional[bool] = None,
    ) -> int:
        """Determine where to insert a pipeline component based on the before/after/first/last values.

        Args:
            before (str): Name or index of the component to insert directly before.
            after (str): Name or index of component to insert directly after.
            first (bool): If True, insert component first in the pipeline.
            last (bool): If True, insert component last in the pipeline.

        Returns:
            (int): The index of the new pipeline component.
        """
        if sum(arg is not None for arg in [before, after, first, last]) >= 2:
            raise LexosException("Only one of before, after, first, last can be set.")
        if last or not any(value is not None for value in [first, before, after]):
            return len(self._components)
        elif first:
            return 0
        elif isinstance(before, str):
            if before not in self.pipes:
                raise LexosException(
                    f"The component name {before} is not in the pipeline."
                )
            return self.pipes.index(before)
        elif isinstance(after, str):
            if after not in self.pipes:
                raise LexosException(
                    f"The component name {after} is not in the pipeline."
                )
            return self.pipes.index(after) + 1
        # We only accept indices referring to components that exist.
        # We can't use isinstance here because bools are instance of int.
        elif type(before) is int:
            if before >= len(self._components) or before < 0:
                raise ValueError(f"Index {before} out of range.")
            return before
        elif type(after) is int:
            if after >= len(self._components) or after < 0:
                raise ValueError(f"Index {after} out of range.")
            return after + 1
        raise ValueError("Invalid combination of before, after, first, last.")

    def add_pipe(
        self,
        components: ScrubberComponent,
        *,
        before: Optional[str | int] = None,
        after: Optional[str | int] = None,
        first: Optional[bool] = None,
        last: Optional[bool] = None,
    ) -> None:
        """Add a component to the scrubber pipeline.

        Args:
            components (ScrubberComponent):
                The component to add to the pipeline. If a string is passed, it is assumed
                to be the name of the component. If a tuple is passed, the first element
                is assumed to be the name of the component and the second element is
                assumed to be a dictionary of options to pass to the component. The method
                also accepts Pipe objects.
            before (str | int): Name or index of the component to insert new
                component directly before.
            after (str | int): Name or index of the component to insert new
                component directly after.
            first (bool): If True, insert component first in the pipeline.
            last (bool): If True, insert component last in the pipeline.
        """
        # Ensure a list of tuples
        components = ensure_list(components)
        pipes = []
        for component in components:
            if isinstance(component, partial):
                name = component.func.__name__
                opts = component.keywords
                pipes.append(Pipe(name=name, opts=opts))
            elif isinstance(component, Pipe):
                pipes.append(component)
            elif isinstance(component, str):
                pipes.append(Pipe(name=component, opts={}))
            elif isinstance(component, tuple):
                name, opts = component
                pipes.append(Pipe(name=name, opts=opts))
            else:
                raise LexosException(
                    "Components must be strings, tuples, functools.partial, or Pipe objects."
                )
        # Create and Pipe objects and insert them into the pipeline
        pipe_index = self._get_pipe_index(before, after, first, last)
        for component in pipes:
            # If component exists, merge options
            if component.name in self.pipes:
                # Find the index of the existing component
                idx = self.pipes.index(component.name)
                instance_opts = self._components[idx].opts
                component.opts = {**instance_opts, **component.opts}
            # Insert the component
            self._components.insert(pipe_index, component)
            pipe_index += 1

    def pipe(
        self,
        texts: Iterable[str],
        *,
        disable: Optional[list[str]] = [],
        component_cfg: Optional[dict[str, dict[str, Any]]] = {},
    ) -> Iterable[str]:
        """Scrub a list of texts with the current pipeline.

        Args:
            texts (Iterable[str]): The text(s) to scrub.
            disable	(Optional[list[str]]): Names of pipeline components to disable.
            component_cfg (Optional[dict[str, dict[str, Any]]]): Optional dictionary of keyword arguments for components, keyed by component names. Defaults to None.

        Yields:
            Iterable: An iterator of scrubbed texts.
        """
        pipes = []
        for pipe in self._components:
            if pipe.name in disable:
                continue
            kwargs = component_cfg.get(pipe.name, None)
            # Allow component_cfg to overwrite the pipe options.
            if kwargs is not None:
                pipe.opts = kwargs
            pipes.append(pipe)
        for text in texts:
            for pipe in pipes:
                text = pipe(text)
            yield text

    def remove_pipe(self, components: str | Iterable[str]) -> None:
        """Remove a component from the scrubber pipeline.

        Args:
            components (str | Iterable[str]): The name of the component to remove from the pipeline.
        """
        self._components = [
            pipe
            for pipe in self._components
            if pipe.name not in ensure_list(components)
        ]

    def reset(self) -> None:
        """Remove all components from the pipeline."""
        self._components = []

    def scrub(self, text: str) -> str:
        """Run a text through the scrubber pipeline.

        Args:
            text (str): The text to scrub.

        Returns:
            str: The scrubbed text.
        """
        for pipe in self._components:
            text = pipe(text)
        return text

pipes: list[Pipe] property ¤

Return a list of the names of the pipeline components.

__init__() -> None ¤

Initialize the Scrubber object.

Source code in lexos/scrubber/scrubber.py 
68
69
70
def __init__(self) -> None:
    """Initialize the Scrubber object."""
    self._components: list[Pipe] = []

add_pipe(components: ScrubberComponent, *, before: Optional[str | int] = None, after: Optional[str | int] = None, first: Optional[bool] = None, last: Optional[bool] = None) -> None ¤

Add a component to the scrubber pipeline.

Parameters:

Name Type Description Default
components ScrubberComponent

The component to add to the pipeline. If a string is passed, it is assumed to be the name of the component. If a tuple is passed, the first element is assumed to be the name of the component and the second element is assumed to be a dictionary of options to pass to the component. The method also accepts Pipe objects.

 required
before str | int

Name or index of the component to insert new component directly before.

 None
after str | int

Name or index of the component to insert new component directly after.

 None
first bool

If True, insert component first in the pipeline.

 None
last bool

If True, insert component last in the pipeline.

 None
Source code in lexos/scrubber/scrubber.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def add_pipe(
    self,
    components: ScrubberComponent,
    *,
    before: Optional[str | int] = None,
    after: Optional[str | int] = None,
    first: Optional[bool] = None,
    last: Optional[bool] = None,
) -> None:
    """Add a component to the scrubber pipeline.

    Args:
        components (ScrubberComponent):
            The component to add to the pipeline. If a string is passed, it is assumed
            to be the name of the component. If a tuple is passed, the first element
            is assumed to be the name of the component and the second element is
            assumed to be a dictionary of options to pass to the component. The method
            also accepts Pipe objects.
        before (str | int): Name or index of the component to insert new
            component directly before.
        after (str | int): Name or index of the component to insert new
            component directly after.
        first (bool): If True, insert component first in the pipeline.
        last (bool): If True, insert component last in the pipeline.
    """
    # Ensure a list of tuples
    components = ensure_list(components)
    pipes = []
    for component in components:
        if isinstance(component, partial):
            name = component.func.__name__
            opts = component.keywords
            pipes.append(Pipe(name=name, opts=opts))
        elif isinstance(component, Pipe):
            pipes.append(component)
        elif isinstance(component, str):
            pipes.append(Pipe(name=component, opts={}))
        elif isinstance(component, tuple):
            name, opts = component
            pipes.append(Pipe(name=name, opts=opts))
        else:
            raise LexosException(
                "Components must be strings, tuples, functools.partial, or Pipe objects."
            )
    # Create and Pipe objects and insert them into the pipeline
    pipe_index = self._get_pipe_index(before, after, first, last)
    for component in pipes:
        # If component exists, merge options
        if component.name in self.pipes:
            # Find the index of the existing component
            idx = self.pipes.index(component.name)
            instance_opts = self._components[idx].opts
            component.opts = {**instance_opts, **component.opts}
        # Insert the component
        self._components.insert(pipe_index, component)
        pipe_index += 1

pipe(texts: Iterable[str], *, disable: Optional[list[str]] = [], component_cfg: Optional[dict[str, dict[str, Any]]] = {}) -> Iterable[str] ¤

Scrub a list of texts with the current pipeline.

Parameters:

Name Type Description Default
texts Iterable[str]

The text(s) to scrub.

 required
disable Optional[list[str]]

Names of pipeline components to disable.

 []
component_cfg Optional[dict[str, dict[str, Any]]]

Optional dictionary of keyword arguments for components, keyed by component names. Defaults to None.

 {}

Yields:

Name Type Description
Iterable Iterable[str]

An iterator of scrubbed texts.
Source code in lexos/scrubber/scrubber.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def pipe(
    self,
    texts: Iterable[str],
    *,
    disable: Optional[list[str]] = [],
    component_cfg: Optional[dict[str, dict[str, Any]]] = {},
) -> Iterable[str]:
    """Scrub a list of texts with the current pipeline.

    Args:
        texts (Iterable[str]): The text(s) to scrub.
        disable	(Optional[list[str]]): Names of pipeline components to disable.
        component_cfg (Optional[dict[str, dict[str, Any]]]): Optional dictionary of keyword arguments for components, keyed by component names. Defaults to None.

    Yields:
        Iterable: An iterator of scrubbed texts.
    """
    pipes = []
    for pipe in self._components:
        if pipe.name in disable:
            continue
        kwargs = component_cfg.get(pipe.name, None)
        # Allow component_cfg to overwrite the pipe options.
        if kwargs is not None:
            pipe.opts = kwargs
        pipes.append(pipe)
    for text in texts:
        for pipe in pipes:
            text = pipe(text)
        yield text

remove_pipe(components: str | Iterable[str]) -> None ¤

Remove a component from the scrubber pipeline.

Parameters:

Name Type Description Default
components str | Iterable[str]

The name of the component to remove from the pipeline.

 required
Source code in lexos/scrubber/scrubber.py 
213
214
215
216
217
218
219
220
221
222
223
def remove_pipe(self, components: str | Iterable[str]) -> None:
    """Remove a component from the scrubber pipeline.

    Args:
        components (str | Iterable[str]): The name of the component to remove from the pipeline.
    """
    self._components = [
        pipe
        for pipe in self._components
        if pipe.name not in ensure_list(components)
    ]

reset() -> None ¤

Remove all components from the pipeline.

Source code in lexos/scrubber/scrubber.py 
225
226
227
def reset(self) -> None:
    """Remove all components from the pipeline."""
    self._components = []

scrub(text: str) -> str ¤

Run a text through the scrubber pipeline.

Parameters:

Name Type Description Default
text str

The text to scrub.

 required

Returns:

Name Type Description
str str

The scrubbed text.
Source code in lexos/scrubber/scrubber.py 
229
230
231
232
233
234
235
236
237
238
239
240
def scrub(self, text: str) -> str:
    """Run a text through the scrubber pipeline.

    Args:
        text (str): The text to scrub.

    Returns:
        str: The scrubbed text.
    """
    for pipe in self._components:
        text = pipe(text)
    return text

__init__() -> None ¤

Initialize the Scrubber object.

Source code in lexos/scrubber/scrubber.py 
68
69
70
def __init__(self) -> None:
    """Initialize the Scrubber object."""
    self._components: list[Pipe] = []

pipes: list[Pipe] property ¤

Return a list of the names of the pipeline components.

_get_pipe_index(before: Optional[str | int] = None, after: Optional[str | int] = None, first: Optional[bool] = None, last: Optional[bool] = None) -> int ¤

Determine where to insert a pipeline component based on the before/after/first/last values.

Parameters:

Name Type Description Default
before str

Name or index of the component to insert directly before.

 None
after str

Name or index of component to insert directly after.

 None
first bool

If True, insert component first in the pipeline.

 None
last bool

If True, insert component last in the pipeline.

 None

Returns:

Type Description
int

The index of the new pipeline component.
Source code in lexos/scrubber/scrubber.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def _get_pipe_index(
    self,
    before: Optional[str | int] = None,
    after: Optional[str | int] = None,
    first: Optional[bool] = None,
    last: Optional[bool] = None,
) -> int:
    """Determine where to insert a pipeline component based on the before/after/first/last values.

    Args:
        before (str): Name or index of the component to insert directly before.
        after (str): Name or index of component to insert directly after.
        first (bool): If True, insert component first in the pipeline.
        last (bool): If True, insert component last in the pipeline.

    Returns:
        (int): The index of the new pipeline component.
    """
    if sum(arg is not None for arg in [before, after, first, last]) >= 2:
        raise LexosException("Only one of before, after, first, last can be set.")
    if last or not any(value is not None for value in [first, before, after]):
        return len(self._components)
    elif first:
        return 0
    elif isinstance(before, str):
        if before not in self.pipes:
            raise LexosException(
                f"The component name {before} is not in the pipeline."
            )
        return self.pipes.index(before)
    elif isinstance(after, str):
        if after not in self.pipes:
            raise LexosException(
                f"The component name {after} is not in the pipeline."
            )
        return self.pipes.index(after) + 1
    # We only accept indices referring to components that exist.
    # We can't use isinstance here because bools are instance of int.
    elif type(before) is int:
        if before >= len(self._components) or before < 0:
            raise ValueError(f"Index {before} out of range.")
        return before
    elif type(after) is int:
        if after >= len(self._components) or after < 0:
            raise ValueError(f"Index {after} out of range.")
        return after + 1
    raise ValueError("Invalid combination of before, after, first, last.")

add_pipe(components: ScrubberComponent, *, before: Optional[str | int] = None, after: Optional[str | int] = None, first: Optional[bool] = None, last: Optional[bool] = None) -> None ¤

Add a component to the scrubber pipeline.

Parameters:

Name Type Description Default
components ScrubberComponent

The component to add to the pipeline. If a string is passed, it is assumed to be the name of the component. If a tuple is passed, the first element is assumed to be the name of the component and the second element is assumed to be a dictionary of options to pass to the component. The method also accepts Pipe objects.

 required
before str | int

Name or index of the component to insert new component directly before.

 None
after str | int

Name or index of the component to insert new component directly after.

 None
first bool

If True, insert component first in the pipeline.

 None
last bool

If True, insert component last in the pipeline.

 None
Source code in lexos/scrubber/scrubber.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def add_pipe(
    self,
    components: ScrubberComponent,
    *,
    before: Optional[str | int] = None,
    after: Optional[str | int] = None,
    first: Optional[bool] = None,
    last: Optional[bool] = None,
) -> None:
    """Add a component to the scrubber pipeline.

    Args:
        components (ScrubberComponent):
            The component to add to the pipeline. If a string is passed, it is assumed
            to be the name of the component. If a tuple is passed, the first element
            is assumed to be the name of the component and the second element is
            assumed to be a dictionary of options to pass to the component. The method
            also accepts Pipe objects.
        before (str | int): Name or index of the component to insert new
            component directly before.
        after (str | int): Name or index of the component to insert new
            component directly after.
        first (bool): If True, insert component first in the pipeline.
        last (bool): If True, insert component last in the pipeline.
    """
    # Ensure a list of tuples
    components = ensure_list(components)
    pipes = []
    for component in components:
        if isinstance(component, partial):
            name = component.func.__name__
            opts = component.keywords
            pipes.append(Pipe(name=name, opts=opts))
        elif isinstance(component, Pipe):
            pipes.append(component)
        elif isinstance(component, str):
            pipes.append(Pipe(name=component, opts={}))
        elif isinstance(component, tuple):
            name, opts = component
            pipes.append(Pipe(name=name, opts=opts))
        else:
            raise LexosException(
                "Components must be strings, tuples, functools.partial, or Pipe objects."
            )
    # Create and Pipe objects and insert them into the pipeline
    pipe_index = self._get_pipe_index(before, after, first, last)
    for component in pipes:
        # If component exists, merge options
        if component.name in self.pipes:
            # Find the index of the existing component
            idx = self.pipes.index(component.name)
            instance_opts = self._components[idx].opts
            component.opts = {**instance_opts, **component.opts}
        # Insert the component
        self._components.insert(pipe_index, component)
        pipe_index += 1

pipe(texts: Iterable[str], *, disable: Optional[list[str]] = [], component_cfg: Optional[dict[str, dict[str, Any]]] = {}) -> Iterable[str] ¤

Scrub a list of texts with the current pipeline.

Parameters:

Name Type Description Default
texts Iterable[str]

The text(s) to scrub.

 required
disable Optional[list[str]]

Names of pipeline components to disable.

 []
component_cfg Optional[dict[str, dict[str, Any]]]

Optional dictionary of keyword arguments for components, keyed by component names. Defaults to None.

 {}

Yields:

Name Type Description
Iterable Iterable[str]

An iterator of scrubbed texts.
Source code in lexos/scrubber/scrubber.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def pipe(
    self,
    texts: Iterable[str],
    *,
    disable: Optional[list[str]] = [],
    component_cfg: Optional[dict[str, dict[str, Any]]] = {},
) -> Iterable[str]:
    """Scrub a list of texts with the current pipeline.

    Args:
        texts (Iterable[str]): The text(s) to scrub.
        disable	(Optional[list[str]]): Names of pipeline components to disable.
        component_cfg (Optional[dict[str, dict[str, Any]]]): Optional dictionary of keyword arguments for components, keyed by component names. Defaults to None.

    Yields:
        Iterable: An iterator of scrubbed texts.
    """
    pipes = []
    for pipe in self._components:
        if pipe.name in disable:
            continue
        kwargs = component_cfg.get(pipe.name, None)
        # Allow component_cfg to overwrite the pipe options.
        if kwargs is not None:
            pipe.opts = kwargs
        pipes.append(pipe)
    for text in texts:
        for pipe in pipes:
            text = pipe(text)
        yield text

remove_pipe(components: str | Iterable[str]) -> None ¤

Remove a component from the scrubber pipeline.

Parameters:

Name Type Description Default
components str | Iterable[str]

The name of the component to remove from the pipeline.

 required
Source code in lexos/scrubber/scrubber.py 
213
214
215
216
217
218
219
220
221
222
223
def remove_pipe(self, components: str | Iterable[str]) -> None:
    """Remove a component from the scrubber pipeline.

    Args:
        components (str | Iterable[str]): The name of the component to remove from the pipeline.
    """
    self._components = [
        pipe
        for pipe in self._components
        if pipe.name not in ensure_list(components)
    ]

reset() -> None ¤

Remove all components from the pipeline.

Source code in lexos/scrubber/scrubber.py 
225
226
227
def reset(self) -> None:
    """Remove all components from the pipeline."""
    self._components = []

scrub(text: str) -> str ¤

Run a text through the scrubber pipeline.

Parameters:

Name Type Description Default
text str

The text to scrub.

 required

Returns:

Name Type Description
str str

The scrubbed text.
Source code in lexos/scrubber/scrubber.py 
229
230
231
232
233
234
235
236
237
238
239
240
def scrub(self, text: str) -> str:
    """Run a text through the scrubber pipeline.

    Args:
        text (str): The text to scrub.

    Returns:
        str: The scrubbed text.
    """
    for pipe in self._components:
        text = pipe(text)
    return text

scrub(text: str, pipeline: PipelineComponents, factory: Optional[catalogue.Registry] = scrubber_components) -> str ¤

Scrub a text with a pipeline of components.

Parameters:

Name Type Description Default
text str

The text to scrub.

 required
pipeline PipelineComponents

An iterable of components. These can be functions, partial functions, Pipe objects, tuple, or strings. If a string is passed, it is assumed to be the name of the component. If a tuple is passed, the first element is assumed to be the name of the component and the second element is assumed to be a dictionary of options to pass.

 required
factory Optional[Registry]

The factory to use to get the components. Defaults to scrubber_components.

 scrubber_components

Returns:

Name Type Description
str str

The scrubbed text.
Source code in lexos/scrubber/scrubber.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def scrub(
    text: str,
    pipeline: PipelineComponents,
    factory: Optional[catalogue.Registry] = scrubber_components,
) -> str:
    """Scrub a text with a pipeline of components.

    Args:
        text (str): The text to scrub.
        pipeline (PipelineComponents): An iterable of components. These can be
            functions, partial functions, Pipe objects, tuple, or strings. If a
            string is passed, it is assumed to be the name of the component.
            If a tuple is passed, the first element is assumed to be the name of
            the component and the second element is assumed to be a dictionary
            of options to pass.
        factory (Optional[catalogue.Registry], optional): The factory to use to get the components. Defaults to scrubber_components.

    Returns:
        str: The scrubbed text.
    """
    for pipe in pipeline:
        if isinstance(pipe, (Callable, partial)):
            text = pipe(text)
        elif isinstance(pipe, tuple):
            func, opts = pipe
            text = func(text, **opts)
        else:
            try:
                func = factory.get(pipe)
                text = func(text)
            except AttributeError as e:
                raise LexosException(e)
            except catalogue.RegistryError as e:
                raise LexosException(e)
    return text