Skip to content

Result

ExecutionResult dataclass

Bases: Generic[T, E]

The unified result container for piping runs.

Attributes:

Name Type Description
result Union[Result[T, E], List[Result[T, E]]]

Either a Result[T, E] (single pipeline) or List[Result[T, E]] (parallel pipelines).
trace Optional[Union[Trace[T, E], Traces[T, E]]]

Optional Trace[T, E] or Traces[T, E] if debug=True.
execution_time float

Elapsed time in seconds.
time_unit str

Always 's'.
Source code in neopipe/result.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
@dataclass
class ExecutionResult(Generic[T, E]):
    """
    The unified result container for piping runs.

    Attributes:
      result: Either a Result[T, E] (single pipeline) or
              List[Result[T, E]] (parallel pipelines).
      trace:  Optional Trace[T, E] or Traces[T, E] if debug=True.
      execution_time: Elapsed time in seconds.
      time_unit: Always 's'.
    """
    result: Union[Result[T, E], List[Result[T, E]]]
    trace: Optional[Union[Trace[T, E], Traces[T, E]]]
    execution_time: float
    time_unit: str = "s"

    def value(self) -> Union[T, List[T]]:
        """
        Extract the inner success value(s). If any entry is Err, raises.
        """
        if isinstance(self.result, list):
            return [r.unwrap() for r in self.result]
        return self.result.unwrap()

    def unwrap(self) -> Union[Result[T, E], List[Result[T, E]]]:
        if self.is_ok():
            return self.result
        raise UnwrapError(f"Called unwrap on Err: {self.result}")

    def is_ok(self) -> bool:
        if isinstance(self.result, list):
            return all(r.is_ok() for r in self.result)
        return self.result.is_ok()

    def is_err(self) -> bool:
        if isinstance(self.result, list):
            return any(r.is_err() for r in self.result)
        return self.result.is_err()


    def __len__(self) -> int:

        if isinstance(self.result, list):
            return len(self.result)
        return 1

    def __repr__(self) -> str:
        base = (
            f"ExecutionResult(result={self.result!r}, "
            f"execution_time={self.execution_time:.3f}{self.time_unit}"
        )
        if self.trace is not None:
            base += f", trace={self.trace!r}"
        base += ")"
        return base

    def __str__(self) -> str:
        return self.__repr__()

value()

Extract the inner success value(s). If any entry is Err, raises.

Source code in neopipe/result.py 
397
398
399
400
401
402
403
def value(self) -> Union[T, List[T]]:
    """
    Extract the inner success value(s). If any entry is Err, raises.
    """
    if isinstance(self.result, list):
        return [r.unwrap() for r in self.result]
    return self.result.unwrap()

Result dataclass

Bases: Generic[T, E]

A Rust-style Result type for monadic error handling in Python.

Source code in neopipe/result.py 
 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
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
241
242
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
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
309
310
311
312
313
314
315
316
317
318
319
320
321
@dataclass(frozen=True)
class Result(Generic[T, E]):
    """A Rust-style Result type for monadic error handling in Python."""

    _is_ok: bool
    _value: Union[T, E]

    @staticmethod
    def Ok(value: T) -> Self:
        """
        Create a Result representing a successful value.

        Args:
            value: The success value.

        Returns:
            Result[T, E]: An Ok variant.
        """
        return Result(True, value)

    @staticmethod
    def Err(error: E) -> Self:
        """
        Create a Result representing an error.

        Args:
            error: The error value.

        Returns:
            Result[T, E]: An Err variant.
        """
        return Result(False, error)

    def is_ok(self) -> bool:
        """
        Check if the result is Ok.

        Returns:
            bool: True if Ok, False otherwise.
        """
        return self._is_ok

    def is_err(self) -> bool:
        """
        Check if the result is Err.

        Returns:
            bool: True if Err, False otherwise.
        """
        return not self._is_ok

    def ok(self) -> Union[T, None]:
        """
        Get the success value if available.

        Returns:
            T | None: The Ok value or None.
        """
        return self._value if self._is_ok else None

    def err(self) -> Union[E, None]:
        """
        Get the error value if available.

        Returns:
            E | None: The Err value or None.
        """
        return self._value if not self._is_ok else None

    def map(self, op: Callable[[T], U]) -> Result[U, E]:
        """
        Apply a function to the Ok value.

        Args:
            op: A function that transforms the success value.

        Returns:
            Result[U, E]: A new Result with transformed success or the original error.
        """
        if self._is_ok:
            return Result.Ok(op(self._value))  # type: ignore
        return Result.Err(self._value)  # type: ignore

    async def map_async(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]:
        """
        Asynchronously apply a function to the Ok value.

        Args:
            op: An async function that transforms the success value.

        Returns:
            Result[U, E]: A new Result with transformed success or the original error.
        """
        if self._is_ok:
            return Result.Ok(await op(self._value))  # type: ignore
        return Result.Err(self._value)  # type: ignore

    def map_err(self, op: Callable[[E], U]) -> Result[T, U]:
        """
        Apply a function to the Err value.

        Args:
            op: A function that transforms the error value.

        Returns:
            Result[T, U]: A new Result with transformed error or the original success.
        """
        if self._is_ok:
            return Result.Ok(self._value)  # type: ignore
        return Result.Err(op(self._value))  # type: ignore

    async def map_err_async(self, op: Callable[[E], Awaitable[U]]) -> Result[T, U]:
        """
        Asynchronously apply a function to the Err value.

        Args:
            op: An async function that transforms the error value.

        Returns:
            Result[T, U]: A new Result with transformed error or the original success.
        """
        if self._is_ok:
            return Result.Ok(self._value)  # type: ignore
        return Result.Err(await op(self._value))  # type: ignore

    def and_then(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]:
        """
        Chain another Result-returning function if current is Ok.

        Args:
            op: A function that takes a success value and returns a new Result.

        Returns:
            Result[U, E]: The new chained result, or the original error.
        """
        if self._is_ok:
            return op(self._value)  # type: ignore
        return Result.Err(self._value)  # type: ignore

    async def and_then_async(
        self, op: Callable[[T], Awaitable[Result[U, E]]]
    ) -> Result[U, E]:
        """
        Asynchronously chain another Result-returning function if current is Ok.

        Args:
            op: An async function that takes a success value and returns a new Result.

        Returns:
            Result[U, E]: The new chained result, or the original error.
        """
        if self._is_ok:
            return await op(self._value)  # type: ignore
        return Result.Err(self._value)  # type: ignore

    def unwrap(self) -> T:
        """
        Extract the success value or raise an error.

        Returns:
            T: The Ok value.

        Raises:
            UnwrapError: If the result is Err.
        """
        if self._is_ok:
            return self._value  # type: ignore
        raise UnwrapError(f"Called unwrap on Err: {self._value}")

    def unwrap_err(self) -> E:
        """
        Extract the error value or raise an error.

        Returns:
            E: The Err value.

        Raises:
            UnwrapError: If the result is Ok.
        """
        if not self._is_ok:
            return self._value  # type: ignore
        raise UnwrapError(f"Called unwrap_err on Ok: {self._value}")

    def unwrap_or(self, default: T) -> T:
        """
        Return the success value or a default.

        Args:
            default: The fallback value.

        Returns:
            T: The Ok value or the default.
        """
        return self._value if self._is_ok else default  # type: ignore

    def unwrap_or_else(self, op: Callable[[E], T]) -> T:
        """
        Return the success value or a value generated from the error.

        Args:
            op: A function that maps the error to a fallback value.

        Returns:
            T: The Ok value or a fallback derived from the error.
        """
        return self._value if self._is_ok else op(self._value)  # type: ignore

    def err_or(self, default: E) -> E:
        """
        Return the error value or a default.

        Args:
            default: The fallback error value.

        Returns:
            E: The Err value or the default.
        """
        return self._value if not self._is_ok else default  # type: ignore

    def err_or_else(self, op: Callable[[T], E]) -> E:
        """
        Return the error value or a value generated from the success value.

        Args:
            op: A function that maps the success value to a fallback error.

        Returns:
            E: The Err value or a fallback derived from the success value.
        """
        return self._value if not self._is_ok else op(self._value)  # type: ignore

    def expect(self, msg: str) -> T:
        """
        Extract the success value or raise with a custom message.

        Args:
            msg: The message to include in the exception.

        Returns:
            T: The Ok value.

        Raises:
            UnwrapError: If the result is Err.
        """
        if self._is_ok:
            return self._value  # type: ignore
        raise UnwrapError(f"{msg}: {self._value}")

    def expect_err(self, msg: str) -> E:
        """
        Extract the error value or raise with a custom message.

        Args:
            msg: The message to include in the exception.

        Returns:
            E: The Err value.

        Raises:
            UnwrapError: If the result is Ok.
        """
        if not self._is_ok:
            return self._value  # type: ignore
        raise UnwrapError(f"{msg}: {self._value}")

    def match(self, ok_fn: Callable[[T], U], err_fn: Callable[[E], U]) -> U:
        """
        Pattern match to handle both Ok and Err branches.

        Args:
            ok_fn: Function to handle Ok.
            err_fn: Function to handle Err.

        Returns:
            U: Result of executing the appropriate handler.
        """
        if self._is_ok:
            return ok_fn(self._value)  # type: ignore
        return err_fn(self._value)  # type: ignore

    def to_dict(self) -> dict:
        """Converts the Result to a dictionary.

        Returns:
            dict: The Result as a dictionary
        """
        return asdict(self)

    def to_json(self) -> str:
        """Converts the Result to a JSON string.

        Returns:
            str: The Result as a JSON string
        """
        return json.dumps(self.to_dict())

    def __repr__(self) -> str:
        """
        Return a string representation of the Result.

        Returns:
            str: Ok(value) or Err(error).
        """
        variant = "Ok" if self._is_ok else "Err"
        return f"{variant}({self._value!r})"

Err(error) staticmethod

Create a Result representing an error.

Parameters:

Name Type Description Default
error E

The error value.

 required

Returns:

Type Description
Self

Result[T, E]: An Err variant.
Source code in neopipe/result.py 
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def Err(error: E) -> Self:
    """
    Create a Result representing an error.

    Args:
        error: The error value.

    Returns:
        Result[T, E]: An Err variant.
    """
    return Result(False, error)

Ok(value) staticmethod

Create a Result representing a successful value.

Parameters:

Name Type Description Default
value T

The success value.

 required

Returns:

Type Description
Self

Result[T, E]: An Ok variant.
Source code in neopipe/result.py 
24
25
26
27
28
29
30
31
32
33
34
35
@staticmethod
def Ok(value: T) -> Self:
    """
    Create a Result representing a successful value.

    Args:
        value: The success value.

    Returns:
        Result[T, E]: An Ok variant.
    """
    return Result(True, value)

__repr__()

Return a string representation of the Result.

Returns:

Name Type Description
str str

Ok(value) or Err(error).
Source code in neopipe/result.py 
313
314
315
316
317
318
319
320
321
def __repr__(self) -> str:
    """
    Return a string representation of the Result.

    Returns:
        str: Ok(value) or Err(error).
    """
    variant = "Ok" if self._is_ok else "Err"
    return f"{variant}({self._value!r})"

and_then(op)

Chain another Result-returning function if current is Ok.

Parameters:

Name Type Description Default
op Callable[[T], Result[U, E]]

A function that takes a success value and returns a new Result.

 required

Returns:

Type Description
Result[U, E]

Result[U, E]: The new chained result, or the original error.
Source code in neopipe/result.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
def and_then(self, op: Callable[[T], Result[U, E]]) -> Result[U, E]:
    """
    Chain another Result-returning function if current is Ok.

    Args:
        op: A function that takes a success value and returns a new Result.

    Returns:
        Result[U, E]: The new chained result, or the original error.
    """
    if self._is_ok:
        return op(self._value)  # type: ignore
    return Result.Err(self._value)  # type: ignore

and_then_async(op) async

Asynchronously chain another Result-returning function if current is Ok.

Parameters:

Name Type Description Default
op Callable[[T], Awaitable[Result[U, E]]]

An async function that takes a success value and returns a new Result.

 required

Returns:

Type Description
Result[U, E]

Result[U, E]: The new chained result, or the original error.
Source code in neopipe/result.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def and_then_async(
    self, op: Callable[[T], Awaitable[Result[U, E]]]
) -> Result[U, E]:
    """
    Asynchronously chain another Result-returning function if current is Ok.

    Args:
        op: An async function that takes a success value and returns a new Result.

    Returns:
        Result[U, E]: The new chained result, or the original error.
    """
    if self._is_ok:
        return await op(self._value)  # type: ignore
    return Result.Err(self._value)  # type: ignore

err()

Get the error value if available.

Returns:

Type Description
Union[E, None]

E | None: The Err value or None.
Source code in neopipe/result.py 
77
78
79
80
81
82
83
84
def err(self) -> Union[E, None]:
    """
    Get the error value if available.

    Returns:
        E | None: The Err value or None.
    """
    return self._value if not self._is_ok else None

err_or(default)

Return the error value or a default.

Parameters:

Name Type Description Default
default E

The fallback error value.

 required

Returns:

Name Type Description
E E

The Err value or the default.
Source code in neopipe/result.py 
224
225
226
227
228
229
230
231
232
233
234
def err_or(self, default: E) -> E:
    """
    Return the error value or a default.

    Args:
        default: The fallback error value.

    Returns:
        E: The Err value or the default.
    """
    return self._value if not self._is_ok else default  # type: ignore

err_or_else(op)

Return the error value or a value generated from the success value.

Parameters:

Name Type Description Default
op Callable[[T], E]

A function that maps the success value to a fallback error.

 required

Returns:

Name Type Description
E E

The Err value or a fallback derived from the success value.
Source code in neopipe/result.py 
236
237
238
239
240
241
242
243
244
245
246
def err_or_else(self, op: Callable[[T], E]) -> E:
    """
    Return the error value or a value generated from the success value.

    Args:
        op: A function that maps the success value to a fallback error.

    Returns:
        E: The Err value or a fallback derived from the success value.
    """
    return self._value if not self._is_ok else op(self._value)  # type: ignore

expect(msg)

Extract the success value or raise with a custom message.

Parameters:

Name Type Description Default
msg str

The message to include in the exception.

 required

Returns:

Name Type Description
T T

The Ok value.

Raises:

Type Description
UnwrapError

If the result is Err.
Source code in neopipe/result.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def expect(self, msg: str) -> T:
    """
    Extract the success value or raise with a custom message.

    Args:
        msg: The message to include in the exception.

    Returns:
        T: The Ok value.

    Raises:
        UnwrapError: If the result is Err.
    """
    if self._is_ok:
        return self._value  # type: ignore
    raise UnwrapError(f"{msg}: {self._value}")

expect_err(msg)

Extract the error value or raise with a custom message.

Parameters:

Name Type Description Default
msg str

The message to include in the exception.

 required

Returns:

Name Type Description
E E

The Err value.

Raises:

Type Description
UnwrapError

If the result is Ok.
Source code in neopipe/result.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def expect_err(self, msg: str) -> E:
    """
    Extract the error value or raise with a custom message.

    Args:
        msg: The message to include in the exception.

    Returns:
        E: The Err value.

    Raises:
        UnwrapError: If the result is Ok.
    """
    if not self._is_ok:
        return self._value  # type: ignore
    raise UnwrapError(f"{msg}: {self._value}")

is_err()

Check if the result is Err.

Returns:

Name Type Description
bool bool

True if Err, False otherwise.
Source code in neopipe/result.py 
59
60
61
62
63
64
65
66
def is_err(self) -> bool:
    """
    Check if the result is Err.

    Returns:
        bool: True if Err, False otherwise.
    """
    return not self._is_ok

is_ok()

Check if the result is Ok.

Returns:

Name Type Description
bool bool

True if Ok, False otherwise.
Source code in neopipe/result.py 
50
51
52
53
54
55
56
57
def is_ok(self) -> bool:
    """
    Check if the result is Ok.

    Returns:
        bool: True if Ok, False otherwise.
    """
    return self._is_ok

map(op)

Apply a function to the Ok value.

Parameters:

Name Type Description Default
op Callable[[T], U]

A function that transforms the success value.

 required

Returns:

Type Description
Result[U, E]

Result[U, E]: A new Result with transformed success or the original error.
Source code in neopipe/result.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
def map(self, op: Callable[[T], U]) -> Result[U, E]:
    """
    Apply a function to the Ok value.

    Args:
        op: A function that transforms the success value.

    Returns:
        Result[U, E]: A new Result with transformed success or the original error.
    """
    if self._is_ok:
        return Result.Ok(op(self._value))  # type: ignore
    return Result.Err(self._value)  # type: ignore

map_async(op) async

Asynchronously apply a function to the Ok value.

Parameters:

Name Type Description Default
op Callable[[T], Awaitable[U]]

An async function that transforms the success value.

 required

Returns:

Type Description
Result[U, E]

Result[U, E]: A new Result with transformed success or the original error.
Source code in neopipe/result.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
async def map_async(self, op: Callable[[T], Awaitable[U]]) -> Result[U, E]:
    """
    Asynchronously apply a function to the Ok value.

    Args:
        op: An async function that transforms the success value.

    Returns:
        Result[U, E]: A new Result with transformed success or the original error.
    """
    if self._is_ok:
        return Result.Ok(await op(self._value))  # type: ignore
    return Result.Err(self._value)  # type: ignore

map_err(op)

Apply a function to the Err value.

Parameters:

Name Type Description Default
op Callable[[E], U]

A function that transforms the error value.

 required

Returns:

Type Description
Result[T, U]

Result[T, U]: A new Result with transformed error or the original success.
Source code in neopipe/result.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
def map_err(self, op: Callable[[E], U]) -> Result[T, U]:
    """
    Apply a function to the Err value.

    Args:
        op: A function that transforms the error value.

    Returns:
        Result[T, U]: A new Result with transformed error or the original success.
    """
    if self._is_ok:
        return Result.Ok(self._value)  # type: ignore
    return Result.Err(op(self._value))  # type: ignore

map_err_async(op) async

Asynchronously apply a function to the Err value.

Parameters:

Name Type Description Default
op Callable[[E], Awaitable[U]]

An async function that transforms the error value.

 required

Returns:

Type Description
Result[T, U]

Result[T, U]: A new Result with transformed error or the original success.
Source code in neopipe/result.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
async def map_err_async(self, op: Callable[[E], Awaitable[U]]) -> Result[T, U]:
    """
    Asynchronously apply a function to the Err value.

    Args:
        op: An async function that transforms the error value.

    Returns:
        Result[T, U]: A new Result with transformed error or the original success.
    """
    if self._is_ok:
        return Result.Ok(self._value)  # type: ignore
    return Result.Err(await op(self._value))  # type: ignore

match(ok_fn, err_fn)

Pattern match to handle both Ok and Err branches.

Parameters:

Name Type Description Default
ok_fn Callable[[T], U]

Function to handle Ok.

 required
err_fn Callable[[E], U]

Function to handle Err.

 required

Returns:

Name Type Description
U U

Result of executing the appropriate handler.
Source code in neopipe/result.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def match(self, ok_fn: Callable[[T], U], err_fn: Callable[[E], U]) -> U:
    """
    Pattern match to handle both Ok and Err branches.

    Args:
        ok_fn: Function to handle Ok.
        err_fn: Function to handle Err.

    Returns:
        U: Result of executing the appropriate handler.
    """
    if self._is_ok:
        return ok_fn(self._value)  # type: ignore
    return err_fn(self._value)  # type: ignore

ok()

Get the success value if available.

Returns:

Type Description
Union[T, None]

T | None: The Ok value or None.
Source code in neopipe/result.py 
68
69
70
71
72
73
74
75
def ok(self) -> Union[T, None]:
    """
    Get the success value if available.

    Returns:
        T | None: The Ok value or None.
    """
    return self._value if self._is_ok else None

to_dict()

Converts the Result to a dictionary.

Returns:

Name Type Description
dict dict

The Result as a dictionary
Source code in neopipe/result.py 
297
298
299
300
301
302
303
def to_dict(self) -> dict:
    """Converts the Result to a dictionary.

    Returns:
        dict: The Result as a dictionary
    """
    return asdict(self)

to_json()

Converts the Result to a JSON string.

Returns:

Name Type Description
str str

The Result as a JSON string
Source code in neopipe/result.py 
305
306
307
308
309
310
311
def to_json(self) -> str:
    """Converts the Result to a JSON string.

    Returns:
        str: The Result as a JSON string
    """
    return json.dumps(self.to_dict())

unwrap()

Extract the success value or raise an error.

Returns:

Name Type Description
T T

The Ok value.

Raises:

Type Description
UnwrapError

If the result is Err.
Source code in neopipe/result.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
def unwrap(self) -> T:
    """
    Extract the success value or raise an error.

    Returns:
        T: The Ok value.

    Raises:
        UnwrapError: If the result is Err.
    """
    if self._is_ok:
        return self._value  # type: ignore
    raise UnwrapError(f"Called unwrap on Err: {self._value}")

unwrap_err()

Extract the error value or raise an error.

Returns:

Name Type Description
E E

The Err value.

Raises:

Type Description
UnwrapError

If the result is Ok.
Source code in neopipe/result.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
def unwrap_err(self) -> E:
    """
    Extract the error value or raise an error.

    Returns:
        E: The Err value.

    Raises:
        UnwrapError: If the result is Ok.
    """
    if not self._is_ok:
        return self._value  # type: ignore
    raise UnwrapError(f"Called unwrap_err on Ok: {self._value}")

unwrap_or(default)

Return the success value or a default.

Parameters:

Name Type Description Default
default T

The fallback value.

 required

Returns:

Name Type Description
T T

The Ok value or the default.
Source code in neopipe/result.py 
200
201
202
203
204
205
206
207
208
209
210
def unwrap_or(self, default: T) -> T:
    """
    Return the success value or a default.

    Args:
        default: The fallback value.

    Returns:
        T: The Ok value or the default.
    """
    return self._value if self._is_ok else default  # type: ignore

unwrap_or_else(op)

Return the success value or a value generated from the error.

Parameters:

Name Type Description Default
op Callable[[E], T]

A function that maps the error to a fallback value.

 required

Returns:

Name Type Description
T T

The Ok value or a fallback derived from the error.
Source code in neopipe/result.py 
212
213
214
215
216
217
218
219
220
221
222
def unwrap_or_else(self, op: Callable[[E], T]) -> T:
    """
    Return the success value or a value generated from the error.

    Args:
        op: A function that maps the error to a fallback value.

    Returns:
        T: The Ok value or a fallback derived from the error.
    """
    return self._value if self._is_ok else op(self._value)  # type: ignore

Trace dataclass

Bases: Generic[T, E]

A sequential trace of one pipeline: steps is a list of (task_name, Result[T, E]).

Source code in neopipe/result.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
@dataclass
class Trace(Generic[T, E]):
    """
    A sequential trace of one pipeline:
    steps is a list of (task_name, Result[T, E]).
    """
    steps: List[Tuple[str, Result[T, E]]]

    def __getitem__(self, index: int) -> Tuple[str, Result[T, E]]:
        """Get a step by index."""
        return self.steps[index]

    def __iter__(self):
        """Iterate over the steps."""
        return iter(self.steps)

    def __len__(self) -> int:
        return len(self.steps)

    def __repr__(self):
        return f"Trace(steps={self.steps!r})"

__getitem__(index)

Get a step by index.

Source code in neopipe/result.py 
342
343
344
def __getitem__(self, index: int) -> Tuple[str, Result[T, E]]:
    """Get a step by index."""
    return self.steps[index]

__iter__()

Iterate over the steps.

Source code in neopipe/result.py 
346
347
348
def __iter__(self):
    """Iterate over the steps."""
    return iter(self.steps)

Traces dataclass

Bases: Generic[T, E]

A collection of per-pipeline traces. pipelines is a list of Trace[T, E].

Source code in neopipe/result.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
@dataclass
class Traces(Generic[T, E]):
    """
    A collection of per-pipeline traces.
    pipelines is a list of Trace[T, E].
    """
    pipelines: List[Trace[T, E]]

    def __getitem__(self, index: int) -> Trace[T, E]:
        """Get a trace by index."""
        return self.pipelines[index]

    def __iter__(self):
        """Iterate over the traces."""
        return iter(self.pipelines)

    def __len__(self) -> int:
        return len(self.pipelines)

    def __repr__(self):
        return f"Traces(pipelines={self.pipelines!r})"

__getitem__(index)

Get a trace by index.

Source code in neopipe/result.py 
365
366
367
def __getitem__(self, index: int) -> Trace[T, E]:
    """Get a trace by index."""
    return self.pipelines[index]

__iter__()

Iterate over the traces.

Source code in neopipe/result.py 
369
370
371
def __iter__(self):
    """Iterate over the traces."""
    return iter(self.pipelines)

UnwrapError

Bases: Exception

Raised when unwrap is called on an Err value.

Source code in neopipe/result.py 
12
13
14
class UnwrapError(Exception):
    """Raised when unwrap is called on an Err value."""
    pass

Err(error)

Creates an Err Result with the given error.

Source code in neopipe/result.py 
329
330
331
def Err(error: E) -> Result[None, E]:
    """Creates an Err Result with the given error."""
    return Result(False, error)

Ok(value)

Creates an Ok Result with the given value.

Source code in neopipe/result.py 
324
325
326
def Ok(value: T) -> Result[T, None]:
    """Creates an Ok Result with the given value."""
    return Result(True, value)