Skip to content

Battery (DRY)

The Battery module provides a DRY (Don't Repeat Yourself) pattern for sharing and reusing pipe configurations across your application. Instead of duplicating the same type, conditions, and matches arguments in every Pipeline, you define a BatteryUnit once and sink() it wherever you need it.

Core Concepts

Class Responsibility
BatteryUnit Stores a single pipe configuration
Battery A centralized registry of pre-built and custom BatteryUnit instances

BatteryUnit

A BatteryUnit is a container for a reusable pipe configuration. It exposes a single method, sink(), which returns the stored configuration - optionally merged with field-specific overrides.

sink() - Consuming a Unit

The sink() method is the primary way to use a BatteryUnit. It returns a PipeConfig dictionary ready to be unpacked into a Pipeline.

from pipeline.battery import Battery

# Use a built-in unit as-is
Pipeline(
    uuid=Battery.UUID.sink()
)

# Override a specific key only for this field
Pipeline(
    email=Battery.use_email().sink(optional=True)
)

Non-Destructive Overrides

sink() merges the overrides without modifying the original BatteryUnit. The same unit can be safely reused in multiple pipelines with different overrides.

Battery Registry

Battery.add() - Registering Custom Units

Register your own units at application startup to make them available globally:

from pipeline import Pipe
from pipeline.battery import Battery

# Register custom unit
Battery.add(
    name="Username",
    type=str,
    conditions={
        Pipe.Condition.MinLength: 3,
        Pipe.Condition.MaxLength: 20
    },
    matches={Pipe.Match.Text.Alphanumeric: None},
    transform={Pipe.Transform.Lowercase: None}
)

Battery.get() - Retrieving Units by Name

# Use it in a pipeline
Pipeline(
    username=Battery.get("Username").sink()
)

Real-World Example

from pipeline import Pipeline, Pipe
from pipeline.battery import Battery

# Register application-wide units once (e.g., in app startup)
Battery.add(
    name="ProductSKU",
    type=str,
    conditions={Pipe.Condition.ExactLength: 8},
    matches={Pipe.Match.Text.Alphanumeric: None},
    transform={Pipe.Transform.Uppercase: None}
)

# Reuse across multiple endpoints without duplication
create_order_pipeline = Pipeline(
    product_sku=Battery.get("ProductSKU").sink(),
    customer_email=Battery.use_email().sink(),
    limit=Battery.use_limit(max_number=50).sink(optional=True),
    offset=Battery.use_offset().sink(optional=True)
)

update_order_pipeline = Pipeline(
    product_sku=Battery.get("ProductSKU").sink(optional=True),
    customer_email=Battery.use_email().sink(optional=True)
)

Technical Reference

A reusable container for a single pipe configuration.

Stores a pre-defined set of pipe arguments that can be retrieved and optionally extended at the point of use via the sink() method.

Source code in pipeline/battery/unit.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class BatteryUnit:
    """
    A reusable container for a single pipe configuration.

    Stores a pre-defined set of pipe arguments that can be retrieved and optionally extended at
    the point of use via the `sink()` method.
    """
    def __init__(self, **pipe_config: Unpack[PipeConfig]) -> None:
        """
        Initializes the BatteryUnit with a fixed pipe configuration.

        Args:
            **pipe_config (PipeConfig): The pipe configuration to store.
        """
        self.pipe_config: PipeConfig = pipe_config

    def sink(self, **update: Unpack[PipeUpdateConfig]) -> PipeConfig:
        """
        Returns the stored pipe configuration, optionally merged with overrides.

        This method is the primary way to consume a BatteryUnit. It returns a
        `PipeConfig` dictionary ready to be passed to a `Pipe` or `Pipeline`.   
        Any keyword arguments provided in `update` will overwrite the
        corresponding keys in the stored configuration, allowing for
        field-specific customization without modifying the original unit.

        Args:
            **update (PipeUpdateConfig): Optional overrides to merge into the stored configuration.

        Returns:
            PipeConfig: A merged dictionary of the stored configuration and any provided overrides.
        """
        if update is None:
            return self.pipe_config

        return {**self.pipe_config, **update}

__init__(**pipe_config)

Initializes the BatteryUnit with a fixed pipe configuration.

Parameters:

Name Type Description Default
**pipe_config PipeConfig

The pipe configuration to store.

 {}
Source code in pipeline/battery/unit.py 
13
14
15
16
17
18
19
20
def __init__(self, **pipe_config: Unpack[PipeConfig]) -> None:
    """
    Initializes the BatteryUnit with a fixed pipe configuration.

    Args:
        **pipe_config (PipeConfig): The pipe configuration to store.
    """
    self.pipe_config: PipeConfig = pipe_config

sink(**update)

Returns the stored pipe configuration, optionally merged with overrides.

This method is the primary way to consume a BatteryUnit. It returns a PipeConfig dictionary ready to be passed to a Pipe or Pipeline.
Any keyword arguments provided in update will overwrite the corresponding keys in the stored configuration, allowing for field-specific customization without modifying the original unit.

Parameters:

Name Type Description Default
**update PipeUpdateConfig

Optional overrides to merge into the stored configuration.

 {}

Returns:

Name Type Description
PipeConfig PipeConfig

A merged dictionary of the stored configuration and any provided overrides.
Source code in pipeline/battery/unit.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def sink(self, **update: Unpack[PipeUpdateConfig]) -> PipeConfig:
    """
    Returns the stored pipe configuration, optionally merged with overrides.

    This method is the primary way to consume a BatteryUnit. It returns a
    `PipeConfig` dictionary ready to be passed to a `Pipe` or `Pipeline`.   
    Any keyword arguments provided in `update` will overwrite the
    corresponding keys in the stored configuration, allowing for
    field-specific customization without modifying the original unit.

    Args:
        **update (PipeUpdateConfig): Optional overrides to merge into the stored configuration.

    Returns:
        PipeConfig: A merged dictionary of the stored configuration and any provided overrides.
    """
    if update is None:
        return self.pipe_config

    return {**self.pipe_config, **update}

A centralized registry of reusable pipe configurations for DRY validation.

Battery serves as a collection of pre-built BatteryUnit instances that represent common validation patterns (e.g., UUID, email, pagination limits). These units can be registered statically as class attributes or added dynamically at runtime, and then consumed anywhere in the application via the sink() method.

Source code in pipeline/battery/battery.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 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
 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
 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
class Battery:
    """
    A centralized registry of reusable pipe configurations for DRY validation.

    Battery serves as a collection of pre-built `BatteryUnit` instances that
    represent common validation patterns (e.g., UUID, email, pagination limits).
    These units can be registered statically as class attributes or added
    dynamically at runtime, and then consumed anywhere in the application
    via the `sink()` method.
    """
    UUID = BatteryUnit(
        type=str,
        conditions={Condition.ExactLength: 36},
        matches={Match.Format.UUID: None}
    )
    """A pre-built unit that validates a UUID v4 string."""

    @classmethod
    def add(cls, name: str, **pipe_config: Unpack[PipeConfig]) -> None:
        """
        Registers a new BatteryUnit as a class attribute.

        Creates a `BatteryUnit` from the provided pipe configuration and
        assigns it to the class under the given name. This allows dynamically
        extending the Battery registry at application startup.

        Args:
            name (str): The attribute name under which the unit will be stored on the class.
            **pipe_config (PipeConfig): The pipe configuration for the new unit.
        """
        setattr(cls, name, BatteryUnit(**pipe_config))

    @classmethod
    def get(cls, name: str) -> BatteryUnit:
        """
        Retrieves a registered BatteryUnit by name.

        Args:
            name (str): The attribute name of the unit to retrieve.

        Returns:
            BatteryUnit: The BatteryUnit instance registered under the given name.

        Raises:
            AttributeError: If no unit with the given name has been registered on the class.
        """
        return getattr(cls, name)

    @staticmethod
    def use_email(min_length: int = 6, max_length: int = 64) -> BatteryUnit:
        """
        Creates a BatteryUnit for email address validation.

        Produces a unit that validates a string value as a well-formed email
        address within a configurable length range.

        Args:
            min_length (int): The minimum allowed length of the email address.
                Defaults to 6.
            max_length (int): The maximum allowed length of the email address.
                Defaults to 64.

        Returns:
            BatteryUnit: A unit for email validation.
        """
        return BatteryUnit(
            type=str,
            conditions={
                Condition.MinLength: min_length,
                Condition.MaxLength: max_length
            },
            matches={Match.Format.Email: None}
        )

    @staticmethod
    def use_limit(min_number: int = 1, max_number: int = 1000) -> BatteryUnit:
        """
        Creates a BatteryUnit for pagination limit validation.

        Produces a unit that validates an integer value representing the
        number of items to return per page, within a configurable range.

        Args:
            min_number (int): The minimum allowed limit value. Defaults to 1.
            max_number (int): The maximum allowed limit value. Defaults to 1000.

        Returns:
            BatteryUnit: A unit for pagination limit validation.
        """
        return BatteryUnit(
            type=int,
            conditions={
                Condition.MinNumber: min_number,
                Condition.MaxNumber: max_number
            }
        )

    @staticmethod
    def use_offset(min_number: int = 0, max_number: int = 1000) -> BatteryUnit:
        """
        Creates a BatteryUnit for pagination offset validation.

        Produces a unit that validates an integer value representing the
        starting position of a result set, within a configurable range.
        Differs from `use_limit` only in its default `min_number` of 0,
        which allows an offset of zero as a valid starting point.

        Args:
            min_number (int): The minimum allowed offset value. Defaults to 0.
            max_number (int): The maximum allowed offset value. Defaults to 1000.

        Returns:
            BatteryUnit: A unit for pagination offset validation.
        """
        return BatteryUnit(
            type=int,
            conditions={
                Condition.MinNumber: min_number,
                Condition.MaxNumber: max_number
            }
        )

UUID = BatteryUnit(type=str, conditions={Condition.ExactLength: 36}, matches={Match.Format.UUID: None}) class-attribute instance-attribute

A pre-built unit that validates a UUID v4 string.

add(name, **pipe_config) classmethod

Registers a new BatteryUnit as a class attribute.

Creates a BatteryUnit from the provided pipe configuration and assigns it to the class under the given name. This allows dynamically extending the Battery registry at application startup.

Parameters:

Name Type Description Default
name str

The attribute name under which the unit will be stored on the class.

 required
**pipe_config PipeConfig

The pipe configuration for the new unit.

 {}
Source code in pipeline/battery/battery.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@classmethod
def add(cls, name: str, **pipe_config: Unpack[PipeConfig]) -> None:
    """
    Registers a new BatteryUnit as a class attribute.

    Creates a `BatteryUnit` from the provided pipe configuration and
    assigns it to the class under the given name. This allows dynamically
    extending the Battery registry at application startup.

    Args:
        name (str): The attribute name under which the unit will be stored on the class.
        **pipe_config (PipeConfig): The pipe configuration for the new unit.
    """
    setattr(cls, name, BatteryUnit(**pipe_config))

get(name) classmethod

Retrieves a registered BatteryUnit by name.

Parameters:

Name Type Description Default
name str

The attribute name of the unit to retrieve.

 required

Returns:

Name Type Description
BatteryUnit BatteryUnit

The BatteryUnit instance registered under the given name.

Raises:

Type Description
AttributeError

If no unit with the given name has been registered on the class.
Source code in pipeline/battery/battery.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@classmethod
def get(cls, name: str) -> BatteryUnit:
    """
    Retrieves a registered BatteryUnit by name.

    Args:
        name (str): The attribute name of the unit to retrieve.

    Returns:
        BatteryUnit: The BatteryUnit instance registered under the given name.

    Raises:
        AttributeError: If no unit with the given name has been registered on the class.
    """
    return getattr(cls, name)

use_email(min_length=6, max_length=64) staticmethod

Creates a BatteryUnit for email address validation.

Produces a unit that validates a string value as a well-formed email address within a configurable length range.

Parameters:

Name Type Description Default
min_length int

The minimum allowed length of the email address. Defaults to 6.

 6
max_length int

The maximum allowed length of the email address. Defaults to 64.

 64

Returns:

Name Type Description
BatteryUnit BatteryUnit

A unit for email validation.
Source code in pipeline/battery/battery.py 
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
@staticmethod
def use_email(min_length: int = 6, max_length: int = 64) -> BatteryUnit:
    """
    Creates a BatteryUnit for email address validation.

    Produces a unit that validates a string value as a well-formed email
    address within a configurable length range.

    Args:
        min_length (int): The minimum allowed length of the email address.
            Defaults to 6.
        max_length (int): The maximum allowed length of the email address.
            Defaults to 64.

    Returns:
        BatteryUnit: A unit for email validation.
    """
    return BatteryUnit(
        type=str,
        conditions={
            Condition.MinLength: min_length,
            Condition.MaxLength: max_length
        },
        matches={Match.Format.Email: None}
    )

use_limit(min_number=1, max_number=1000) staticmethod

Creates a BatteryUnit for pagination limit validation.

Produces a unit that validates an integer value representing the number of items to return per page, within a configurable range.

Parameters:

Name Type Description Default
min_number int

The minimum allowed limit value. Defaults to 1.

 1
max_number int

The maximum allowed limit value. Defaults to 1000.

 1000

Returns:

Name Type Description
BatteryUnit BatteryUnit

A unit for pagination limit validation.
Source code in pipeline/battery/battery.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@staticmethod
def use_limit(min_number: int = 1, max_number: int = 1000) -> BatteryUnit:
    """
    Creates a BatteryUnit for pagination limit validation.

    Produces a unit that validates an integer value representing the
    number of items to return per page, within a configurable range.

    Args:
        min_number (int): The minimum allowed limit value. Defaults to 1.
        max_number (int): The maximum allowed limit value. Defaults to 1000.

    Returns:
        BatteryUnit: A unit for pagination limit validation.
    """
    return BatteryUnit(
        type=int,
        conditions={
            Condition.MinNumber: min_number,
            Condition.MaxNumber: max_number
        }
    )

use_offset(min_number=0, max_number=1000) staticmethod

Creates a BatteryUnit for pagination offset validation.

Produces a unit that validates an integer value representing the starting position of a result set, within a configurable range. Differs from use_limit only in its default min_number of 0, which allows an offset of zero as a valid starting point.

Parameters:

Name Type Description Default
min_number int

The minimum allowed offset value. Defaults to 0.

 0
max_number int

The maximum allowed offset value. Defaults to 1000.

 1000

Returns:

Name Type Description
BatteryUnit BatteryUnit

A unit for pagination offset validation.
Source code in pipeline/battery/battery.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@staticmethod
def use_offset(min_number: int = 0, max_number: int = 1000) -> BatteryUnit:
    """
    Creates a BatteryUnit for pagination offset validation.

    Produces a unit that validates an integer value representing the
    starting position of a result set, within a configurable range.
    Differs from `use_limit` only in its default `min_number` of 0,
    which allows an offset of zero as a valid starting point.

    Args:
        min_number (int): The minimum allowed offset value. Defaults to 0.
        max_number (int): The maximum allowed offset value. Defaults to 1000.

    Returns:
        BatteryUnit: A unit for pagination offset validation.
    """
    return BatteryUnit(
        type=int,
        conditions={
            Condition.MinNumber: min_number,
            Condition.MaxNumber: max_number
        }
    )