Skip to content

Validation API Reference

Data filters and validation decorators.

Filters

Base Filter

dspu.validation.filters.Filter

Bases: ABC

Base class for all filters.

Filters are transformation functions that take an input value and return a modified version. They can be chained together using FilterChain.

Example

class MyFilter(Filter): ... def apply(self, value: Any) -> Any: ... return str(value).upper()

f = MyFilter() f("hello") 'HELLO'

Functions

apply abstractmethod 

apply(value: Any) -> Any

Apply the filter transformation to a value.

Parameters:

Name Type Description Default
value Any

The input value to transform

 required

Returns:

Type Description
Any

The transformed value
Source code in src/dspu/validation/filters.py 
28
29
30
31
32
33
34
35
36
37
@abstractmethod
def apply(self, value: Any) -> Any:
    """Apply the filter transformation to a value.

    Args:
        value: The input value to transform

    Returns:
        The transformed value
    """

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

Built-in Filters

dspu.validation.filters.StripWhitespaceFilter

Bases: Filter

Remove leading and trailing whitespace from strings.

Example

f = StripWhitespaceFilter() f(" hello ") 'hello' f("\n\t test \n") 'test'

Functions

apply 

apply(value: Any) -> Any

Strip whitespace from string value.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

String with whitespace stripped
Source code in src/dspu/validation/filters.py 
142
143
144
145
146
147
148
149
150
151
def apply(self, value: Any) -> Any:
    """Strip whitespace from string value.

    Args:
        value: Input value (converted to string)

    Returns:
        String with whitespace stripped
    """
    return str(value).strip()

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.LowercaseFilter

Bases: Filter

Convert strings to lowercase.

Example

f = LowercaseFilter() f("HELLO World") 'hello world'

Functions

apply 

apply(value: Any) -> Any

Convert value to lowercase string.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

Lowercase string
Source code in src/dspu/validation/filters.py 
163
164
165
166
167
168
169
170
171
172
def apply(self, value: Any) -> Any:
    """Convert value to lowercase string.

    Args:
        value: Input value (converted to string)

    Returns:
        Lowercase string
    """
    return str(value).lower()

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.UppercaseFilter

Bases: Filter

Convert strings to uppercase.

Example

f = UppercaseFilter() f("hello world") 'HELLO WORLD'

Functions

apply 

apply(value: Any) -> Any

Convert value to uppercase string.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

Uppercase string
Source code in src/dspu/validation/filters.py 
184
185
186
187
188
189
190
191
192
193
def apply(self, value: Any) -> Any:
    """Convert value to uppercase string.

    Args:
        value: Input value (converted to string)

    Returns:
        Uppercase string
    """
    return str(value).upper()

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.TruncateFilter 

TruncateFilter(max_length: int, suffix: str = '')

Bases: Filter

Truncate strings to a maximum length.

Attributes:

Name Type Description
max_length

Maximum length of output string
suffix

Optional suffix to add to truncated strings
Example

f = TruncateFilter(max_length=10) f("Hello World!") 'Hello Worl'

f = TruncateFilter(max_length=10, suffix="...") f("Hello World!") 'Hello W...'

Initialize truncate filter.

Parameters:

Name Type Description Default
max_length int

Maximum length of output string

 required
suffix str

Optional suffix to add when truncating (default: "")

 ''

Raises:

Type Description
ValueError

If max_length is not positive
Source code in src/dspu/validation/filters.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def __init__(self, max_length: int, suffix: str = "") -> None:
    """Initialize truncate filter.

    Args:
        max_length: Maximum length of output string
        suffix: Optional suffix to add when truncating (default: "")

    Raises:
        ValueError: If max_length is not positive
    """
    if max_length <= 0:
        msg = "max_length must be positive"
        raise ValueError(msg)

    if suffix and len(suffix) >= max_length:
        msg = "suffix length must be less than max_length"
        raise ValueError(msg)

    self.max_length = max_length
    self.suffix = suffix

Functions

apply 

apply(value: Any) -> Any

Truncate string to maximum length.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

Truncated string
Source code in src/dspu/validation/filters.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def apply(self, value: Any) -> Any:
    """Truncate string to maximum length.

    Args:
        value: Input value (converted to string)

    Returns:
        Truncated string
    """
    text = str(value)
    if len(text) <= self.max_length:
        return text

    if self.suffix:
        truncate_to = self.max_length - len(self.suffix)
        return text[:truncate_to] + self.suffix

    return text[: self.max_length]

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.RemoveSpecialCharsFilter 

RemoveSpecialCharsFilter(
    allowed_chars: set[str] | None = None,
    replacement: str = "",
)

Bases: Filter

Remove special characters from strings.

Keeps only alphanumeric characters and specified allowed characters.

Attributes:

Name Type Description
allowed_chars

Set of additional characters to allow (besides alphanumeric)
replacement

Character to replace special characters with
Example

f = RemoveSpecialCharsFilter() f("hello@world!") 'helloworld'

f = RemoveSpecialCharsFilter(allowed_chars={" ", "-"}) f("hello-world!") 'hello-world'

f = RemoveSpecialCharsFilter(replacement="") f("hello@world!") 'hello_world'

Initialize remove special chars filter.

Parameters:

Name Type Description Default
allowed_chars set[str] | None

Set of additional characters to allow (default: None)

 None
replacement str

Character to replace special characters with (default: "")

 ''
Source code in src/dspu/validation/filters.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
def __init__(
    self,
    allowed_chars: set[str] | None = None,
    replacement: str = "",
) -> None:
    """Initialize remove special chars filter.

    Args:
        allowed_chars: Set of additional characters to allow (default: None)
        replacement: Character to replace special characters with (default: "")
    """
    self.allowed_chars = allowed_chars or set()
    self.replacement = replacement

Functions

apply 

apply(value: Any) -> Any

Remove special characters from string.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

String with special characters removed or replaced
Source code in src/dspu/validation/filters.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def apply(self, value: Any) -> Any:
    """Remove special characters from string.

    Args:
        value: Input value (converted to string)

    Returns:
        String with special characters removed or replaced
    """
    text = str(value)
    result: list[str] = []

    for char in text:
        if char.isalnum() or char in self.allowed_chars:
            result.append(char)
        elif self.replacement:
            result.append(self.replacement)

    return "".join(result)

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.EmailNormalizationFilter

Bases: Filter

Normalize email addresses.

Converts email to lowercase and removes dots from Gmail addresses (as Gmail ignores dots in the username part).

Example

f = EmailNormalizationFilter() f("John.Doe@Example.COM") 'johndoe@example.com'

f("test.user+tag@gmail.com") 'testuser@gmail.com'

Functions

apply 

apply(value: Any) -> Any

Normalize email address.

Parameters:

Name Type Description Default
value Any

Email address string

 required

Returns:

Type Description
Any

Normalized email address

Raises:

Type Description
ValueError

If value is not a valid email format
Source code in src/dspu/validation/filters.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def apply(self, value: Any) -> Any:
    """Normalize email address.

    Args:
        value: Email address string

    Returns:
        Normalized email address

    Raises:
        ValueError: If value is not a valid email format
    """
    email = str(value).strip().lower()

    # Basic email validation
    if "@" not in email or email.count("@") != 1:
        msg = f"Invalid email format: {value}"
        raise ValueError(msg)

    username, domain = email.split("@")

    # Remove dots from Gmail addresses
    if domain in ("gmail.com", "googlemail.com"):
        username = username.replace(".", "")

    # Remove + tags (common email aliasing)
    if "+" in username:
        username = username.split("+")[0]

    return f"{username}@{domain}"

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.SlugifyFilter 

SlugifyFilter(separator: str = '-', lowercase: bool = True)

Bases: Filter

Convert strings to URL-friendly slugs.

Converts text to lowercase, replaces spaces and special characters with hyphens, and removes consecutive hyphens.

Attributes:

Name Type Description
separator

Character to use as separator (default: "-")
lowercase

Whether to convert to lowercase (default: True)
Example

f = SlugifyFilter() f("Hello World!") 'hello-world'

f(" Multiple Spaces ") 'multiple-spaces'

f = SlugifyFilter(separator="_", lowercase=False) f("Hello World!") 'Hello_World'

Initialize slugify filter.

Parameters:

Name Type Description Default
separator str

Character to use as separator (default: "-")

 '-'
lowercase bool

Whether to convert to lowercase (default: True)

 True
Source code in src/dspu/validation/filters.py 
382
383
384
385
386
387
388
389
390
def __init__(self, separator: str = "-", lowercase: bool = True) -> None:
    """Initialize slugify filter.

    Args:
        separator: Character to use as separator (default: "-")
        lowercase: Whether to convert to lowercase (default: True)
    """
    self.separator = separator
    self.lowercase = lowercase

Functions

apply 

apply(value: Any) -> Any

Convert string to slug.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

URL-friendly slug
Source code in src/dspu/validation/filters.py 
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
def apply(self, value: Any) -> Any:
    """Convert string to slug.

    Args:
        value: Input value (converted to string)

    Returns:
        URL-friendly slug
    """
    text = str(value)

    # Convert to lowercase if requested
    if self.lowercase:
        text = text.lower()

    # Replace whitespace and special chars with separator
    text = re.sub(r"[^\w\s-]", "", text)
    text = re.sub(r"[-\s]+", self.separator, text)

    # Remove leading/trailing separators
    return text.strip(self.separator)

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

dspu.validation.filters.RegexReplaceFilter 

RegexReplaceFilter(
    pattern: str, replacement: str, count: int = 0
)

Bases: Filter

Replace text matching a regex pattern.

Attributes:

Name Type Description
pattern

Regex pattern to match
replacement

String to replace matches with
count

Maximum number of replacements (0 = unlimited)
Example

f = RegexReplaceFilter(r"\d+", "X") f("I have 123 apples and 456 oranges") 'I have X apples and X oranges'

f = RegexReplaceFilter(r"\d+", "X", count=1) f("I have 123 apples and 456 oranges") 'I have X apples and 456 oranges'

Initialize regex replace filter.

Parameters:

Name Type Description Default
pattern str

Regex pattern to match

 required
replacement str

String to replace matches with

 required
count int

Maximum number of replacements (default: 0 = unlimited)

 0
Source code in src/dspu/validation/filters.py 
433
434
435
436
437
438
439
440
441
442
443
def __init__(self, pattern: str, replacement: str, count: int = 0) -> None:
    """Initialize regex replace filter.

    Args:
        pattern: Regex pattern to match
        replacement: String to replace matches with
        count: Maximum number of replacements (default: 0 = unlimited)
    """
    self.pattern = re.compile(pattern)
    self.replacement = replacement
    self.count = count

Functions

apply 

apply(value: Any) -> Any

Replace text matching regex pattern.

Parameters:

Name Type Description Default
value Any

Input value (converted to string)

 required

Returns:

Type Description
Any

String with replacements applied
Source code in src/dspu/validation/filters.py 
445
446
447
448
449
450
451
452
453
454
def apply(self, value: Any) -> Any:
    """Replace text matching regex pattern.

    Args:
        value: Input value (converted to string)

    Returns:
        String with replacements applied
    """
    return self.pattern.sub(self.replacement, str(value), count=self.count)

then 

then(other: Filter) -> FilterChain

Chain this filter with another.

Parameters:

Name Type Description Default
other Filter

The filter to chain after this one

 required

Returns:

Type Description
FilterChain

A FilterChain combining both filters
Example

strip = StripWhitespaceFilter() lower = LowercaseFilter() chain = strip.then(lower) chain(" HELLO ") 'hello'

Source code in src/dspu/validation/filters.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def then(self, other: "Filter") -> "FilterChain":
    """Chain this filter with another.

    Args:
        other: The filter to chain after this one

    Returns:
        A FilterChain combining both filters

    Example:
        >>> strip = StripWhitespaceFilter()
        >>> lower = LowercaseFilter()
        >>> chain = strip.then(lower)
        >>> chain("  HELLO  ")
        'hello'
    """
    return FilterChain([self, other])

Filter Chain

dspu.validation.filters.FilterChain 

FilterChain(filters: list[Filter])

Bases: Filter

Chain multiple filters together.

Applies filters in sequence, passing the output of each filter as the input to the next.

Attributes:

Name Type Description
filters

List of filters to apply in order
Example

chain = FilterChain([ ... StripWhitespaceFilter(), ... LowercaseFilter(), ... TruncateFilter(max_length=10) ... ]) chain(" HELLO WORLD ") 'hello worl'

Initialize filter chain.

Parameters:

Name Type Description Default
filters list[Filter]

List of filters to apply in sequence

 required
Source code in src/dspu/validation/filters.py 
88
89
90
91
92
93
94
95
96
97
98
def __init__(self, filters: list[Filter]) -> None:
    """Initialize filter chain.

    Args:
        filters: List of filters to apply in sequence
    """
    if not filters:
        msg = "FilterChain must contain at least one filter"
        raise ValueError(msg)

    self.filters = filters

Functions

apply 

apply(value: Any) -> Any

Apply all filters in sequence.

Parameters:

Name Type Description Default
value Any

The input value to transform

 required

Returns:

Type Description
Any

The transformed value after all filters applied
Source code in src/dspu/validation/filters.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def apply(self, value: Any) -> Any:
    """Apply all filters in sequence.

    Args:
        value: The input value to transform

    Returns:
        The transformed value after all filters applied
    """
    result = value
    for filter_func in self.filters:
        result = filter_func.apply(result)
    return result

then 

then(other: Filter) -> FilterChain

Add another filter to the chain.

Parameters:

Name Type Description Default
other Filter

The filter to add to the end of the chain

 required

Returns:

Type Description
FilterChain

A new FilterChain with the additional filter
Source code in src/dspu/validation/filters.py 
114
115
116
117
118
119
120
121
122
123
def then(self, other: Filter) -> "FilterChain":
    """Add another filter to the chain.

    Args:
        other: The filter to add to the end of the chain

    Returns:
        A new FilterChain with the additional filter
    """
    return FilterChain([*self.filters, other])

Pydantic Integration

dspu.validation.validators.pydantic_filter_validator 

pydantic_filter_validator(
    filter_func: Filter | FilterChain,
) -> Callable[[Any], Any]

Create a Pydantic field validator from a filter.

This allows using dspu filters as Pydantic field validators.

Parameters:

Name Type Description Default
filter_func Filter | FilterChain

Filter or FilterChain to use for validation

 required

Returns:

Type Description
Callable[[Any], Any]

Pydantic validator function

Raises:

Type Description
ImportError

If Pydantic is not installed
Example

from pydantic import BaseModel, field_validator

class User(BaseModel): ... email: str ... name: str ... ... _validate_email = field_validator("email", mode="before")( ... pydantic_filter_validator(EmailNormalizationFilter()) ... ) ... _validate_name = field_validator("name", mode="before")( ... pydantic_filter_validator( ... FilterChain([StripWhitespaceFilter(), TruncateFilter(50)]) ... ) ... )

user = User(email="John.Doe@GMAIL.COM", name=" John Doe ") user.email 'johndoe@gmail.com' user.name 'John Doe'

Source code in src/dspu/validation/validators.py 
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
def pydantic_filter_validator(
    filter_func: Filter | FilterChain,
) -> Callable[[Any], Any]:
    """Create a Pydantic field validator from a filter.

    This allows using dspu filters as Pydantic field validators.

    Args:
        filter_func: Filter or FilterChain to use for validation

    Returns:
        Pydantic validator function

    Raises:
        ImportError: If Pydantic is not installed

    Example:
        >>> from pydantic import BaseModel, field_validator
        >>>
        >>> class User(BaseModel):
        ...     email: str
        ...     name: str
        ...
        ...     _validate_email = field_validator("email", mode="before")(
        ...         pydantic_filter_validator(EmailNormalizationFilter())
        ...     )
        ...     _validate_name = field_validator("name", mode="before")(
        ...         pydantic_filter_validator(
        ...             FilterChain([StripWhitespaceFilter(), TruncateFilter(50)])
        ...         )
        ...     )
        >>>
        >>> user = User(email="John.Doe@GMAIL.COM", name="  John Doe  ")
        >>> user.email
        'johndoe@gmail.com'
        >>> user.name
        'John Doe'
    """
    if not PYDANTIC_AVAILABLE:
        msg = (
            "Pydantic is required for pydantic_filter_validator. Install with: pip install pydantic"
        )
        raise ImportError(msg)

    def validator(value: Any) -> Any:
        """Apply filter to field value."""
        try:
            return filter_func.apply(value)
        except Exception as e:
            msg = f"Validation failed: {e}"
            raise ValueError(msg) from e

    return validator

dspu.validation.validators.FilteredModel

Bases: BaseModel

Base Pydantic model with filter support.

Subclass this to automatically apply filters to fields using a _filters class attribute.

Example

class User(FilteredModel): ... email: str ... name: str ... ... _filters = { ... "email": EmailNormalizationFilter(), ... "name": FilterChain([ ... StripWhitespaceFilter(), ... TruncateFilter(max_length=50) ... ]) ... }

user = User(email="John.Doe@GMAIL.COM", name=" John Doe ") user.email 'johndoe@gmail.com'

Functions

apply_filters_wrap classmethod 

apply_filters_wrap(values: Any, handler: Any) -> Any

Apply filters before validation using wrap mode.

Parameters:

Name Type Description Default
values Any

Input values

 required
handler Any

Validation handler

 required

Returns:

Type Description
Any

Validated model instance

Raises:

Type Description
ImportError

If Pydantic is not installed
Source code in src/dspu/validation/validators.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@model_validator(mode="wrap")  # type: ignore[misc]
@classmethod
def apply_filters_wrap(cls, values: Any, handler: Any) -> Any:
    """Apply filters before validation using wrap mode.

    Args:
        values: Input values
        handler: Validation handler

    Returns:
        Validated model instance

    Raises:
        ImportError: If Pydantic is not installed
    """
    if not PYDANTIC_AVAILABLE:
        msg = "Pydantic is required for FilteredModel. Install with: pip install pydantic"
        raise ImportError(msg)

    # Look for _filters in the class hierarchy
    # In Pydantic 2, _filters is stored in __private_attributes__
    filters = None
    for base_cls in cls.__mro__:
        # Check __private_attributes__ first (Pydantic 2.x)
        if (
            hasattr(base_cls, "__private_attributes__")
            and "_filters" in base_cls.__private_attributes__
        ):
            filters_attr = base_cls.__private_attributes__["_filters"]
            if hasattr(filters_attr, "default"):
                filters = filters_attr.default
            break
        # Fallback to __dict__ for non-Pydantic usage
        if "_filters" in base_cls.__dict__:
            filters_attr = base_cls.__dict__["_filters"]
            filters = filters_attr.default if hasattr(filters_attr, "default") else filters_attr
            break

    # Apply filters if found and input is dict
    if filters and isinstance(values, dict):
        filtered_values = values.copy()
        for field_name, filter_func in filters.items():
            if field_name in filtered_values:
                try:
                    filtered_values[field_name] = filter_func.apply(filtered_values[field_name])
                except Exception as e:
                    msg = f"Filter validation failed for field '{field_name}': {e}"
                    raise ValueError(msg) from e

        # Call handler with filtered values
        return handler(filtered_values)

    # Call handler with original values
    return handler(values)

Decorators

dspu.validation.validators.validate_field 

validate_field(
    *filters: Filter, field_name: str | None = None
) -> Callable[[F], F]

Decorator to validate and transform a specific function argument.

Parameters:

Name Type Description Default
*filters Filter

Filters to apply to the field value

 ()
field_name str | None

Name of the parameter to validate (if None, validates first param)

 None

Returns:

Type Description
Callable[[F], F]

Decorated function that applies filters to the specified field
Example

@validate_field(StripWhitespaceFilter(), LowercaseFilter()) ... def process_email(email: str) -> str: ... return email

process_email(" USER@EXAMPLE.COM ") 'user@example.com'

Source code in src/dspu/validation/validators.py 
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
63
64
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
def validate_field(
    *filters: Filter,
    field_name: str | None = None,
) -> Callable[[F], F]:
    """Decorator to validate and transform a specific function argument.

    Args:
        *filters: Filters to apply to the field value
        field_name: Name of the parameter to validate (if None, validates first param)

    Returns:
        Decorated function that applies filters to the specified field

    Example:
        >>> @validate_field(StripWhitespaceFilter(), LowercaseFilter())
        ... def process_email(email: str) -> str:
        ...     return email
        >>>
        >>> process_email("  USER@EXAMPLE.COM  ")
        'user@example.com'
    """
    if not filters:
        msg = "At least one filter must be provided"
        raise ValueError(msg)

    filter_chain = FilterChain(list(filters)) if len(filters) > 1 else filters[0]

    def decorator(func: F) -> F:
        sig = inspect.signature(func)
        params = list(sig.parameters.keys())

        # Determine which parameter to validate
        if field_name is None:
            if not params:
                msg = f"Function {func.__name__} has no parameters to validate"
                raise ValueError(msg)
            target_param = params[0]
        else:
            if field_name not in params:
                msg = f"Parameter '{field_name}' not found in function {func.__name__}"
                raise ValueError(msg)
            target_param = field_name

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            # Convert args to kwargs for easier manipulation
            bound_args = sig.bind(*args, **kwargs)
            bound_args.apply_defaults()

            # Apply filter to target parameter
            if target_param in bound_args.arguments:
                original_value = bound_args.arguments[target_param]
                try:
                    filtered_value = filter_chain.apply(original_value)
                    bound_args.arguments[target_param] = filtered_value
                except Exception as e:
                    msg = f"Validation failed for parameter '{target_param}': {e}"
                    raise ValidationError(msg) from e

            return func(*bound_args.args, **bound_args.kwargs)

        return wrapper  # type: ignore[return-value]

    return decorator

dspu.validation.validators.validate 

validate(
    *,
    input_filters: dict[str, Filter | FilterChain]
    | None = None,
    output_filter: Filter | FilterChain | None = None,
) -> Callable[[F], F]

Decorator to validate function inputs and outputs.

Parameters:

Name Type Description Default
input_filters dict[str, Filter | FilterChain] | None

Dict mapping parameter names to filters

 None
output_filter Filter | FilterChain | None

Filter to apply to function return value

 None

Returns:

Type Description
Callable[[F], F]

Decorated function with validation applied
Example

@validate( ... input_filters={ ... "email": EmailNormalizationFilter(), ... "name": FilterChain([ ... StripWhitespaceFilter(), ... TruncateFilter(max_length=50) ... ]) ... }, ... output_filter=StripWhitespaceFilter() ... ) ... def create_user(email: str, name: str) -> str: ... return f"{name} <{email}>"

create_user("John.Doe@GMAIL.COM", " John Doe ") 'John Doe johndoe@gmail.com'

Source code in src/dspu/validation/validators.py 
 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
def validate(
    *,
    input_filters: dict[str, Filter | FilterChain] | None = None,
    output_filter: Filter | FilterChain | None = None,
) -> Callable[[F], F]:
    """Decorator to validate function inputs and outputs.

    Args:
        input_filters: Dict mapping parameter names to filters
        output_filter: Filter to apply to function return value

    Returns:
        Decorated function with validation applied

    Example:
        >>> @validate(
        ...     input_filters={
        ...         "email": EmailNormalizationFilter(),
        ...         "name": FilterChain([
        ...             StripWhitespaceFilter(),
        ...             TruncateFilter(max_length=50)
        ...         ])
        ...     },
        ...     output_filter=StripWhitespaceFilter()
        ... )
        ... def create_user(email: str, name: str) -> str:
        ...     return f"{name} <{email}>"
        >>>
        >>> create_user("John.Doe@GMAIL.COM", "  John Doe  ")
        'John Doe <johndoe@gmail.com>'
    """

    def decorator(func: F) -> F:
        sig = inspect.signature(func)

        # Validate that input_filters reference valid parameters
        if input_filters:
            for param_name in input_filters:
                if param_name not in sig.parameters:
                    msg = f"Parameter '{param_name}' not found in function {func.__name__}"
                    raise ValueError(msg)

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            # Apply input filters
            if input_filters:
                bound_args = sig.bind(*args, **kwargs)
                bound_args.apply_defaults()

                for param_name, filter_func in input_filters.items():
                    if param_name in bound_args.arguments:
                        original_value = bound_args.arguments[param_name]
                        try:
                            filtered_value = filter_func.apply(original_value)
                            bound_args.arguments[param_name] = filtered_value
                        except Exception as e:
                            msg = f"Validation failed for parameter '{param_name}': {e}"
                            raise ValidationError(msg) from e

                result = func(*bound_args.args, **bound_args.kwargs)
            else:
                result = func(*args, **kwargs)

            # Apply output filter
            if output_filter is not None:
                try:
                    result = output_filter.apply(result)
                except Exception as e:
                    msg = f"Output validation failed: {e}"
                    raise ValidationError(msg) from e

            return result

        return wrapper  # type: ignore[return-value]

    return decorator

Usage

from dspu.validation import StripWhitespaceFilter, LowercaseFilter, EmailNormalizationFilter

# Compose filters
email_filter = (
    StripWhitespaceFilter()
    .then(LowercaseFilter())
    .then(EmailNormalizationFilter())
)

# Apply
clean_email = email_filter("  Alice@GMAIL.COM  ")
# Result: "alice@gmail.com"

# Pydantic integration
from pydantic import BaseModel

class User(BaseModel):
    email: str

    _email_filter = pydantic_filter_validator("email", email_filter)

user = User(email="  ALICE@EXAMPLE.COM  ")
# Automatically cleaned: "alice@example.com"

See Also