xfp
1# Warning : import order matters 2from xfp.utils import curry, curry_method, tupled 3from xfp.xresult import ( 4 Xresult, 5 XRBranch, 6 XresultError, 7 Xeither, 8 Xopt, 9 Xtry, 10) 11from xfp.xlist import Xlist 12from xfp.xiter import Xiter 13from xfp.xdict import Xdict 14 15__all__ = [ 16 "curry", 17 "curry_method", 18 "tupled", 19 "Xiter", 20 "Xlist", 21 "Xresult", 22 "XRBranch", 23 "XresultError", 24 "Xeither", 25 "Xopt", 26 "Xtry", 27 "Xdict", 28]
def
curry(f: F1[typing.Concatenate[~A, ~P], ~T]) -> F1[[~A], typing.Any]:
72def curry(f: F1[Concatenate[A, P], T]) -> F1[[A], Any]: return _curry(f)
def
curry_method( f: F1[typing.Concatenate[~Self, ~A, ~P], ~T]) -> F1[[~Self, ~A], typing.Any]:
100def curry_method(f: F1[Concatenate[Self, A, P], T]) -> F1[[Self, A], Any]: return _curry_method(f)
def
tupled(f: F1[..., Out]) -> F1[[tuple], Out]:
128def tupled[Out](f: F1[..., Out]) -> F1[[tuple], Out]: return _tupled(f)
class
Xiter(typing.Generic[+X]):
16class Xiter(Generic[X]): 17 """Enhance Lists (lazy) with functional behaviors. 18 19 This class provides common behaviors used for declarative programming. 20 21 ### Features 22 23 - Monadic behavior 24 - List proxies or quality of lifes 25 - Iter proxy from (itertools homogene) 26 """ 27 28 @classmethod 29 def cycle[T](cls, c: Iterable[T]) -> "Xiter[T]": 30 "Proxy for itertools.cycle." 31 return Xiter(itertools.cycle(c)) 32 33 @classmethod 34 def repeat[T](cls, x: T) -> "Xiter[T]": 35 "Proxy for itertools.repeat." 36 return Xiter(itertools.repeat(x)) 37 38 def __init__(self, iterable: Iterable[X]) -> None: 39 """Construct an Xiter from an iterable.""" 40 match iterable: 41 case ABCIterable(): 42 self.__iter: Iterator = iter(iterable) 43 case _: 44 raise TypeError("Xiter must be constructed from an iterable") 45 46 def __iter__(self) -> Iterator[X]: 47 """Return an iterable over the underlying data.""" 48 return self.__iter 49 50 def __repr__(self) -> str: 51 """Return the representation of the underlying data""" 52 return repr(self.__iter) 53 54 def __next__(self) -> X: 55 """Return the next element of the iterator. 56 57 Consume this element in the data structure. 58 """ 59 return next(self.__iter) 60 61 def __getitem__(self, i: int) -> X: 62 """Alias for get(i). 63 64 Exists to enable [] syntax 65 """ 66 return self.get(i) 67 68 def takewhile(self, predicate: F1[[X], bool]) -> "Xiter[X]": 69 """Return a new iterator that stops yielding elements when predicate = False. 70 71 Do not consume the original iterator. 72 73 Useful to limit an infinite Xiter with a predicate. 74 75 ### Usage 76 77 ```python 78 from xfp import Xiter 79 import itertools 80 81 until_xiter = ( 82 Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...]) 83 .takewhile(lambda x: x<6) # -> Xiter([0,2,4]) 84 ) 85 ``` 86 """ 87 88 return Xiter(itertools.takewhile(predicate, self.copy().__iter)) 89 90 def copy(self) -> "Xiter[X]": 91 """Return a new Xiter, tee-ed from self. 92 93 Used to make a shallow copy of the iterator, functional style. 94 95 ## Usage 96 97 ```python 98 from xfp import Xiter 99 100 r1 = Xiter(range(10)) 101 r2 = r1.copy() 102 assert next(r1) == 0 103 assert next(r2) == 0 104 105 ``` 106 """ 107 a, b = tee(self) 108 self.__iter = a 109 return Xiter(b) 110 111 def deepcopy(self) -> "Xiter[X]": 112 """Return a new Xiter, with both iterator and elements distincts from self. 113 114 Used to make a deep copy of the iterator, functional style. 115 116 ## Usage 117 118 ```python 119 from xfp import Xiter 120 from dataclasses import dataclass 121 122 @dataclass 123 class A: 124 text: str 125 126 ori = Xiter([A("hello")]) 127 deep_copy = ori.deepcopy() 128 shallow_copy = ori.copy() 129 130 value1 = next(ori) 131 value2 = next(deep_copy) 132 value3 = next(shallow_copy) 133 134 value1.text = "world" # 'ori' is mutated 135 assert value2.text == "hello" # 'deep_copy' is left untouched 136 assert value3.text == "world" # on the contrary, 'shallow_copy' still dependents on 'ori' 137 ``` 138 """ 139 a, b = tee(self) 140 self.__iter = Xiter(a) 141 return Xiter(map(deepcopy, b)) 142 143 def chain[T](self, other: Iterable[T]) -> "Xiter[X | T]": 144 """Proxy for itertools.chain. 145 146 Return a chain object whose `.__next__()` method returns elements from the 147 first iterable until it is exhausted, then elements from the next 148 iterable, until all of the iterables are exhausted. 149 """ 150 return Xiter(itertools.chain(self, other)) 151 152 def get(self, i: int) -> X: 153 """Return the i-th element of the Xiter. 154 155 Does not consume the i-1 first elements, but evaluate them. 156 157 ### Raise 158 159 - IndexError -- if the Xiter is shorter than i 160 """ 161 __copy = self.copy() 162 163 try: 164 for _ in range(i): 165 next(__copy) 166 return next(__copy) 167 except StopIteration: 168 raise IndexError(f"Xiter has less than {i} element(s)") 169 170 def get_fr(self, i: int) -> Xresult[IndexError, X]: 171 """Return the i-th element of the Xiter. 172 173 Does not consume the i-1 first elements, but evaluate them. 174 Wrap the potential error in an Xresult. 175 """ 176 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(i))) 177 178 @deprecated("1.1.0", "2.0.0", details="Use get_fr instead") 179 def get_fx(self, i: int) -> Xresult[IndexError, X]: 180 """Return the i-th element of the Xiter. 181 182 Does not consume the i-1 first elements, but evaluate them. 183 Wrap the potential error in an Xresult. 184 """ 185 return self.get_fr(i) 186 187 def head(self) -> X: 188 """Alias for get(0).""" 189 return self.get(0) 190 191 def head_fr(self) -> Xresult[IndexError, X]: 192 """Alias for get_fr(0).""" 193 return self.get_fr(0) 194 195 @deprecated("1.1.0", "2.0.0", details="Use head_fr instead") 196 def head_fx(self) -> Xresult[IndexError, X]: 197 """Alias for get_fx(0).""" 198 return self.head_fr() 199 200 def tail(self) -> "Xiter[X]": 201 """Return the iterator / its first element. 202 203 ### Raise 204 205 - IndexError -- if the list is empty. 206 """ 207 try: 208 out = self.copy() 209 next(out) 210 return out 211 except StopIteration: 212 raise IndexError("<tail> operation not allowed on empty iterator") 213 214 def tail_fr(self) -> Xresult[IndexError, "Xiter[X]"]: 215 """Return the iterator / its first element. 216 217 Wrap the potential error in an Xresult. 218 """ 219 return cast(Xresult[IndexError, "Xiter[X]"], Xtry.from_unsafe(self.tail)) 220 221 @deprecated("1.1.0", "2.0.0", details="Use tail_fr instead") 222 def tail_fx(self) -> Xresult[IndexError, "Xiter[X]"]: 223 """Return the iterator / its first element. 224 225 Wrap the potential error in an Xresult. 226 """ 227 return self.tail_fr() 228 229 def appended[T](self: "Xiter[T]", el: T) -> "Xiter[T]": 230 """Return a new iterator with el appended. 231 232 After exhaustion of self, the next `next` call will return `el`. 233 """ 234 return self.chain([el]) 235 236 def prepended[T](self: "Xiter[T]", el: T) -> "Xiter[T]": 237 """Return a new iterator with el prepended. 238 239 Before iterating over self, the first `next` call will return `el`. 240 """ 241 return Xiter([el]).chain(self) 242 243 def map[T](self, f: F1[[X], T]) -> "Xiter[T]": 244 """Return a new iterator, with f applied to each future element. 245 246 ### Usage 247 248 ```python 249 from xfp import Xiter 250 251 input = Xiter([1, 2, 3]) 252 assert next(input) == 1 253 f = lambda el: el*el 254 result = input.map(f) 255 assert next(result) == 4 # Xiter([2*2, 3*3]) => 2*2 == 4 256 ``` 257 """ 258 return Xiter(map(f, self.copy())) 259 260 def filter(self, predicate: F1[[X], bool]) -> "Xiter[X]": 261 """Return a new iterator skipping the elements with predicate = False. 262 263 ### Usage 264 265 ```python 266 from xfp import Xiter 267 268 input = Xiter(range(1,5)) 269 predicate = lambda el: el % 2 == 0 270 r1 = input.filter(predicate) 271 # keep only even numbers 272 assert next(r1) == 2 273 assert next(r1) == 4 274 ``` 275 """ 276 return Xiter(filter(predicate, self.copy())) 277 278 def foreach(self, statement: F1[[X], Any]) -> None: 279 """Do the 'statement' procedure once for each element of the iterator. 280 281 Do not consume the original iterator. 282 283 ### Usage 284 285 ```python 286 from xfp import Xiter 287 288 input = Xiter(range(1,4)) 289 statement = lambda el: print(f"This is an element of the range : ${el}") 290 input.foreach(statement) 291 # This is an element of the range : 1 292 # This is an element of the range : 2 293 # This is an element of the range : 3 294 295 input.foreach(statement) # you can reconsume the same iterable 296 # This is an element of the range : 1 297 # This is an element of the range : 2 298 # This is an element of the range : 3 299 300 ``` 301 """ 302 [statement(e) for e in self.copy()] 303 304 def flatten[XS](self: "Xiter[Iterable[XS]]") -> "Xiter[XS]": 305 """Return a new iterator, with each element nested iterated on individually. 306 307 ## Usage 308 309 ```python 310 from xfp import Xiter 311 312 # All the following resulting objects are equivalent to Xiter([1,2,3]) 313 Xiter([1, 2, 3]).flatten() 314 Xiter([[1, 2], [3]]).flatten() 315 Xiter([[1, 2], 3]).flatten() 316 ``` 317 """ 318 319 def result(xi): 320 for el in xi: 321 for inner_el in el: 322 yield inner_el 323 324 return Xiter(result(self.copy())) 325 326 def flat_map[T](self, f: F1[[X], Iterable[T]]) -> "Xiter[T]": 327 """Return the result of map and then flatten. 328 329 Exists as homogenisation with Xresult.flat_map. 330 331 ### Usage 332 333 ```python 334 from xfp import Xiter, Xlist 335 336 Xiter([1, 2, 3]).flat_map(lambda x: Xlist([(x, 4), (x, 5)])) 337 # equivalent to Xiter([(1, 4), (1, 5), (2, 4), (2, 5), (3, 4), (3, 5)]) 338 ``` 339 """ 340 return self.map(f).flatten() 341 342 def fold_left[T](self, zero: T, f: F1[[T, X], T]) -> T: 343 """Return the accumulation of the Xiter elements. 344 345 - Uses a custom accumulator (zero, f) to aggregate the elements of the Xiter 346 - Initialize the accumulator with the zero value 347 - Then from the first to the last element, compute accumulator(n+1) using f, accumulator(n) and self.data[n], such as: 348 accumulator(n+1) = f(accumulator(n), self.data[n]) 349 - Return the last state of the accumulator 350 351 ### Keyword Arguments 352 353 - zero -- initial state of the accumulator 354 - f -- accumulation function, compute the next state of the accumulator 355 356 ### Warnings 357 358 This function falls in infinite loop in the case of infinite iterator. 359 360 ### Usage 361 362 ```python 363 from xfp import Xiter 364 365 assert Xiter([1, 2, 3]).fold_left(0)(lambda x, y: x + y) == 6 366 assert Xiter([1, 2, 3]).fold_left(10)(lambda x, y: x + y) == 16 367 assert Xiter(["1", "2", "3"]).fold_left("")(lambda x, y: x + y) == "123" 368 assert Xiter([]).fold_left(0)(lambda x, y: x + y) == 0 369 ``` 370 """ 371 acc: T = zero 372 for e in self: 373 acc = f(acc, e) 374 return acc 375 376 def fold[T](self, zero: T, f: F1[[T, X], T]) -> T: 377 """Return the accumulation of the Xiter elements. 378 379 Shorthand for fold_left 380 """ 381 return self.fold_left(zero, f) 382 383 def reduce(self, f: F1[[X, X], X]) -> X: 384 """Return the accumulation of the Xiter elements using the first element as the initial state of accumulation. 385 386 ### Raise 387 388 - IndexError -- when the Xiter is empty 389 390 ### Keyword Arguments 391 392 - f -- accumulation function, compute the next state of the accumulator 393 394 ### Warning 395 396 This function falls in infinite loop in the case of infinite iterator. 397 398 ### Usage 399 400 ```python 401 from xfp import Xiter 402 import pytest 403 404 assert Xiter([1, 2, 3]).reduce(lambda x, y: x + y) == 6 405 assert Xiter(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123" 406 with pytest.raises(IndexError): 407 Xiter([]).reduce(lambda x, y: x + y) 408 ``` 409 """ 410 try: 411 h = self.head() 412 except IndexError: 413 raise IndexError("<reduce> operation not allowed on empty list") 414 return self.tail().fold(h, f) 415 416 def reduce_fr(self, f: F1[[X, X], X]) -> Xresult[IndexError, X]: 417 """Return the accumulation of the Xiter elements using the first element as the initial state of accumulation. 418 419 Wrap the potential error in an Xresult. 420 421 ### Keyword Arguments 422 423 - f -- accumulation function, compute the next state of the accumulator 424 425 ### Warning 426 427 This function falls in infinite loop in the case of infinite iterator. 428 429 ### Usage 430 431 ```python 432 from xfp import Xiter, Xtry 433 434 Xiter([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6) 435 Xiter(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123") 436 Xiter([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list")) 437 438 ``` 439 """ 440 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.reduce(f))) 441 442 def min(self, key: F1[[X], _Comparable] = id) -> X: 443 """Return the smallest element of the Xiter given the key criteria. 444 445 ### Raise 446 447 - ValueError -- when the Xiter is empty 448 449 ### Keyword Arguments 450 451 - key (default id) -- the function which extrapolate a sortable from the elements of the list 452 453 ### Warning 454 455 This function falls in infinite loop in the case of infinite iterator. 456 457 ### Usage 458 459 ```python 460 from xfp import Xiter 461 import pytest 462 463 input = Xiter(["ae", "bd", "cc"]) 464 assert input.min() == "ae" 465 assert input.min(lambda x: x[-1]) == "cc" 466 with pytest.raises(IndexError): 467 Xiter([]).min() 468 ``` 469 """ 470 return min(self, key=key) 471 472 def min_fr(self, key: F1[[X], _Comparable] = id) -> Xresult[ValueError, X]: 473 """Return the smallest element of the Xiter given the key criteria. 474 475 Wrap the potential failure in an Wresult 476 477 ### Warning 478 479 This function falls in infinite loop in the case of infinite iterator. 480 ``` 481 """ 482 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.min(key))) 483 484 def max(self, key: F1[[X], _Comparable] = id) -> X: 485 """Return the bigget element of the Xiter given the key criteria. 486 487 ### Raise 488 489 - ValueError -- when the Xiter is empty 490 491 ### Keyword Arguments 492 493 - key (default id) -- the function which extrapolate a sortable from the elements of the list 494 495 ### Warning 496 497 This function falls in infinite loop in the case of infinite iterator. 498 499 ### Usage 500 501 ```python 502 from xfp import Xiter 503 import pytest 504 505 input = Xiter(["ae", "bd", "cc"]) 506 assert input.max() == "cc" 507 assert input.max(lambda x: x[-1]) == "ae" 508 with pytest.raises(IndexError): 509 Xiter([]).max() 510 ``` 511 """ 512 return max(self, key=key) 513 514 def max_fr(self, key: F1[[X], _Comparable] = id) -> Xresult[ValueError, X]: 515 """Return the biggest element of the Xiter given the key criteria. 516 517 Wrap the potential failure in an Wresult 518 519 ### Warning 520 521 This function falls in infinite loop in the case of infinite iterator. 522 ``` 523 """ 524 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.max(key))) 525 526 def take(self, n: int) -> "Xiter[X]": 527 """Return a new iterator limited to the first 'n' elements. 528 Return a copy if the original iterator has less than 'n' elements. 529 Return an empty Xiter if n is negative. 530 531 Do not consume the original iterator. 532 533 ### Usage 534 535 ```python 536 from xfp import Xiter 537 import itertools 538 539 infinite_xiter = Xiter(itertools.repeat(42)) # -> Xiter([42,42,42,...]) 540 until_xiter = infinite_xiter.take(3) # -> Xiter([42,42]) 541 ``` 542 """ 543 544 return Xiter(self.slice(n)) 545 546 def takeuntil(self, predicate: F1[[X], bool]) -> "Xiter[X]": 547 """Return a new iterator that stops yielding elements when predicate = True. 548 549 Do not consume the original iterator. 550 551 Useful to limit an infinite Xiter with a predicate. 552 553 ### Usage 554 555 ```python 556 from xfp import Xiter 557 import itertools 558 559 infinite_xiter = Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...]) 560 until_xiter = infinite_xiter.takeuntil(lambda x: x >=6) # -> Xiter([0,2,4]) 561 ``` 562 """ 563 564 # fixes "error: Cannot use a covariant type variable as a parameter" on lambda x: not predicate(x) 565 @curry2 566 def invert[T](p: F1[[T], bool], x: T) -> bool: 567 return not p(x) 568 569 return self.takewhile(invert(predicate)) 570 571 @overload 572 def slice(self, stop: int | None, /) -> "Xiter[X]": ... 573 574 @overload 575 def slice( 576 self, start: int | None, stop: int | None, step: int | None = 1, / 577 ) -> "Xiter[X]": ... 578 579 def slice(self, *args) -> "Xiter[X]": 580 """Return an new Xiter with selected elements from the Xiter. 581 Works like sequence slicing but does not support negative values 582 for start, stop, or step. 583 584 Do not consume the original iterator. 585 586 If start is zero or None, iteration starts at zero. 587 Otherwise, elements from the iterable are skipped until start is reached. 588 589 If stop is None, iteration continues until the input is exhausted, 590 if at all. Otherwise, it stops at the specified position. 591 592 If step is None, the step defaults to one. 593 Elements are returned consecutively unless step is set higher than 594 one which results in items being skipped. 595 """ 596 __iter_copy = self.copy() 597 598 if len(args) not in (1, 2, 3): 599 raise TypeError( 600 "slice expected from 1 to 3 positional arguments: 'stop' | 'start' 'stop' ['step']" 601 ) 602 603 return Xiter(itertools.islice(__iter_copy, *args)) 604 605 def zip[T](self, other: Iterable[T]) -> "Xiter[tuple[X, T]]": 606 """Zip this iterator with another iterable.""" 607 return Xiter(zip(self.copy(), other)) 608 609 def to_Xlist(self) -> "Xlist[X]": 610 """Return an Xlist being the evaluated version of self. 611 612 Do not consume the original iterator. 613 """ 614 return Xlist(self.copy())
Enhance Lists (lazy) with functional behaviors.
This class provides common behaviors used for declarative programming.
Features
- Monadic behavior
- List proxies or quality of lifes
- Iter proxy from (itertools homogene)
Xiter(iterable: Iterable[+X])
38 def __init__(self, iterable: Iterable[X]) -> None: 39 """Construct an Xiter from an iterable.""" 40 match iterable: 41 case ABCIterable(): 42 self.__iter: Iterator = iter(iterable) 43 case _: 44 raise TypeError("Xiter must be constructed from an iterable")
Construct an Xiter from an iterable.
@classmethod
def
cycle(cls, c: Iterable[T]) -> 'Xiter[T]':
28 @classmethod 29 def cycle[T](cls, c: Iterable[T]) -> "Xiter[T]": 30 "Proxy for itertools.cycle." 31 return Xiter(itertools.cycle(c))
Proxy for itertools.cycle.
@classmethod
def
repeat(cls, x: T) -> 'Xiter[T]':
33 @classmethod 34 def repeat[T](cls, x: T) -> "Xiter[T]": 35 "Proxy for itertools.repeat." 36 return Xiter(itertools.repeat(x))
Proxy for itertools.repeat.
68 def takewhile(self, predicate: F1[[X], bool]) -> "Xiter[X]": 69 """Return a new iterator that stops yielding elements when predicate = False. 70 71 Do not consume the original iterator. 72 73 Useful to limit an infinite Xiter with a predicate. 74 75 ### Usage 76 77 ```python 78 from xfp import Xiter 79 import itertools 80 81 until_xiter = ( 82 Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...]) 83 .takewhile(lambda x: x<6) # -> Xiter([0,2,4]) 84 ) 85 ``` 86 """ 87 88 return Xiter(itertools.takewhile(predicate, self.copy().__iter))
Return a new iterator that stops yielding elements when predicate = False.
Do not consume the original iterator.
Useful to limit an infinite Xiter with a predicate.
Usage
from xfp import Xiter
import itertools
until_xiter = (
Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...])
.takewhile(lambda x: x<6) # -> Xiter([0,2,4])
)
90 def copy(self) -> "Xiter[X]": 91 """Return a new Xiter, tee-ed from self. 92 93 Used to make a shallow copy of the iterator, functional style. 94 95 ## Usage 96 97 ```python 98 from xfp import Xiter 99 100 r1 = Xiter(range(10)) 101 r2 = r1.copy() 102 assert next(r1) == 0 103 assert next(r2) == 0 104 105 ``` 106 """ 107 a, b = tee(self) 108 self.__iter = a 109 return Xiter(b)
Return a new Xiter, tee-ed from self.
Used to make a shallow copy of the iterator, functional style.
Usage
from xfp import Xiter
r1 = Xiter(range(10))
r2 = r1.copy()
assert next(r1) == 0
assert next(r2) == 0
111 def deepcopy(self) -> "Xiter[X]": 112 """Return a new Xiter, with both iterator and elements distincts from self. 113 114 Used to make a deep copy of the iterator, functional style. 115 116 ## Usage 117 118 ```python 119 from xfp import Xiter 120 from dataclasses import dataclass 121 122 @dataclass 123 class A: 124 text: str 125 126 ori = Xiter([A("hello")]) 127 deep_copy = ori.deepcopy() 128 shallow_copy = ori.copy() 129 130 value1 = next(ori) 131 value2 = next(deep_copy) 132 value3 = next(shallow_copy) 133 134 value1.text = "world" # 'ori' is mutated 135 assert value2.text == "hello" # 'deep_copy' is left untouched 136 assert value3.text == "world" # on the contrary, 'shallow_copy' still dependents on 'ori' 137 ``` 138 """ 139 a, b = tee(self) 140 self.__iter = Xiter(a) 141 return Xiter(map(deepcopy, b))
Return a new Xiter, with both iterator and elements distincts from self.
Used to make a deep copy of the iterator, functional style.
Usage
from xfp import Xiter
from dataclasses import dataclass
@dataclass
class A:
text: str
ori = Xiter([A("hello")])
deep_copy = ori.deepcopy()
shallow_copy = ori.copy()
value1 = next(ori)
value2 = next(deep_copy)
value3 = next(shallow_copy)
value1.text = "world" # 'ori' is mutated
assert value2.text == "hello" # 'deep_copy' is left untouched
assert value3.text == "world" # on the contrary, 'shallow_copy' still dependents on 'ori'
def
chain(self, other: Iterable[T]) -> 'Xiter[X | T]':
143 def chain[T](self, other: Iterable[T]) -> "Xiter[X | T]": 144 """Proxy for itertools.chain. 145 146 Return a chain object whose `.__next__()` method returns elements from the 147 first iterable until it is exhausted, then elements from the next 148 iterable, until all of the iterables are exhausted. 149 """ 150 return Xiter(itertools.chain(self, other))
Proxy for itertools.chain.
Return a chain object whose .__next__() method returns elements from the
first iterable until it is exhausted, then elements from the next
iterable, until all of the iterables are exhausted.
def
get(self, i: int) -> +X:
152 def get(self, i: int) -> X: 153 """Return the i-th element of the Xiter. 154 155 Does not consume the i-1 first elements, but evaluate them. 156 157 ### Raise 158 159 - IndexError -- if the Xiter is shorter than i 160 """ 161 __copy = self.copy() 162 163 try: 164 for _ in range(i): 165 next(__copy) 166 return next(__copy) 167 except StopIteration: 168 raise IndexError(f"Xiter has less than {i} element(s)")
Return the i-th element of the Xiter.
Does not consume the i-1 first elements, but evaluate them.
Raise
- IndexError -- if the Xiter is shorter than i
170 def get_fr(self, i: int) -> Xresult[IndexError, X]: 171 """Return the i-th element of the Xiter. 172 173 Does not consume the i-1 first elements, but evaluate them. 174 Wrap the potential error in an Xresult. 175 """ 176 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(i)))
Return the i-th element of the Xiter.
Does not consume the i-1 first elements, but evaluate them. Wrap the potential error in an Xresult.
@deprecated('1.1.0', '2.0.0', details='Use get_fr instead')
def
get_fx(self, i: int) -> Xresult[IndexError, +X]:
178 @deprecated("1.1.0", "2.0.0", details="Use get_fr instead") 179 def get_fx(self, i: int) -> Xresult[IndexError, X]: 180 """Return the i-th element of the Xiter. 181 182 Does not consume the i-1 first elements, but evaluate them. 183 Wrap the potential error in an Xresult. 184 """ 185 return self.get_fr(i)
Return the i-th element of the Xiter.
Does not consume the i-1 first elements, but evaluate them. Wrap the potential error in an Xresult.
Deprecated since version 1.1.0: This will be removed in 2.0.0. Use get_fr instead
191 def head_fr(self) -> Xresult[IndexError, X]: 192 """Alias for get_fr(0).""" 193 return self.get_fr(0)
Alias for get_fr(0).
@deprecated('1.1.0', '2.0.0', details='Use head_fr instead')
def
head_fx(self) -> Xresult[IndexError, +X]:
195 @deprecated("1.1.0", "2.0.0", details="Use head_fr instead") 196 def head_fx(self) -> Xresult[IndexError, X]: 197 """Alias for get_fx(0).""" 198 return self.head_fr()
Alias for get_fx(0).
Deprecated since version 1.1.0: This will be removed in 2.0.0. Use head_fr instead
200 def tail(self) -> "Xiter[X]": 201 """Return the iterator / its first element. 202 203 ### Raise 204 205 - IndexError -- if the list is empty. 206 """ 207 try: 208 out = self.copy() 209 next(out) 210 return out 211 except StopIteration: 212 raise IndexError("<tail> operation not allowed on empty iterator")
Return the iterator / its first element.
Raise
- IndexError -- if the list is empty.
214 def tail_fr(self) -> Xresult[IndexError, "Xiter[X]"]: 215 """Return the iterator / its first element. 216 217 Wrap the potential error in an Xresult. 218 """ 219 return cast(Xresult[IndexError, "Xiter[X]"], Xtry.from_unsafe(self.tail))
Return the iterator / its first element.
Wrap the potential error in an Xresult.
@deprecated('1.1.0', '2.0.0', details='Use tail_fr instead')
def
tail_fx(self) -> Xresult[IndexError, Xiter[+X]]:
221 @deprecated("1.1.0", "2.0.0", details="Use tail_fr instead") 222 def tail_fx(self) -> Xresult[IndexError, "Xiter[X]"]: 223 """Return the iterator / its first element. 224 225 Wrap the potential error in an Xresult. 226 """ 227 return self.tail_fr()
Return the iterator / its first element.
Wrap the potential error in an Xresult.
Deprecated since version 1.1.0: This will be removed in 2.0.0. Use tail_fr instead
def
appended(self: 'Xiter[T]', el: T) -> 'Xiter[T]':
229 def appended[T](self: "Xiter[T]", el: T) -> "Xiter[T]": 230 """Return a new iterator with el appended. 231 232 After exhaustion of self, the next `next` call will return `el`. 233 """ 234 return self.chain([el])
Return a new iterator with el appended.
After exhaustion of self, the next next call will return el.
def
prepended(self: 'Xiter[T]', el: T) -> 'Xiter[T]':
236 def prepended[T](self: "Xiter[T]", el: T) -> "Xiter[T]": 237 """Return a new iterator with el prepended. 238 239 Before iterating over self, the first `next` call will return `el`. 240 """ 241 return Xiter([el]).chain(self)
Return a new iterator with el prepended.
Before iterating over self, the first next call will return el.
def
map(self, f: F1[[+X], T]) -> 'Xiter[T]':
243 def map[T](self, f: F1[[X], T]) -> "Xiter[T]": 244 """Return a new iterator, with f applied to each future element. 245 246 ### Usage 247 248 ```python 249 from xfp import Xiter 250 251 input = Xiter([1, 2, 3]) 252 assert next(input) == 1 253 f = lambda el: el*el 254 result = input.map(f) 255 assert next(result) == 4 # Xiter([2*2, 3*3]) => 2*2 == 4 256 ``` 257 """ 258 return Xiter(map(f, self.copy()))
Return a new iterator, with f applied to each future element.
Usage
from xfp import Xiter
input = Xiter([1, 2, 3])
assert next(input) == 1
f = lambda el: el*el
result = input.map(f)
assert next(result) == 4 # Xiter([2*2, 3*3]) => 2*2 == 4
260 def filter(self, predicate: F1[[X], bool]) -> "Xiter[X]": 261 """Return a new iterator skipping the elements with predicate = False. 262 263 ### Usage 264 265 ```python 266 from xfp import Xiter 267 268 input = Xiter(range(1,5)) 269 predicate = lambda el: el % 2 == 0 270 r1 = input.filter(predicate) 271 # keep only even numbers 272 assert next(r1) == 2 273 assert next(r1) == 4 274 ``` 275 """ 276 return Xiter(filter(predicate, self.copy()))
Return a new iterator skipping the elements with predicate = False.
Usage
from xfp import Xiter
input = Xiter(range(1,5))
predicate = lambda el: el % 2 == 0
r1 = input.filter(predicate)
# keep only even numbers
assert next(r1) == 2
assert next(r1) == 4
def
foreach(self, statement: F1[[+X], typing.Any]) -> None:
278 def foreach(self, statement: F1[[X], Any]) -> None: 279 """Do the 'statement' procedure once for each element of the iterator. 280 281 Do not consume the original iterator. 282 283 ### Usage 284 285 ```python 286 from xfp import Xiter 287 288 input = Xiter(range(1,4)) 289 statement = lambda el: print(f"This is an element of the range : ${el}") 290 input.foreach(statement) 291 # This is an element of the range : 1 292 # This is an element of the range : 2 293 # This is an element of the range : 3 294 295 input.foreach(statement) # you can reconsume the same iterable 296 # This is an element of the range : 1 297 # This is an element of the range : 2 298 # This is an element of the range : 3 299 300 ``` 301 """ 302 [statement(e) for e in self.copy()]
Do the 'statement' procedure once for each element of the iterator.
Do not consume the original iterator.
Usage
from xfp import Xiter
input = Xiter(range(1,4))
statement = lambda el: print(f"This is an element of the range : ${el}")
input.foreach(statement)
# This is an element of the range : 1
# This is an element of the range : 2
# This is an element of the range : 3
input.foreach(statement) # you can reconsume the same iterable
# This is an element of the range : 1
# This is an element of the range : 2
# This is an element of the range : 3
def
flatten(self: 'Xiter[Iterable[XS]]') -> 'Xiter[XS]':
304 def flatten[XS](self: "Xiter[Iterable[XS]]") -> "Xiter[XS]": 305 """Return a new iterator, with each element nested iterated on individually. 306 307 ## Usage 308 309 ```python 310 from xfp import Xiter 311 312 # All the following resulting objects are equivalent to Xiter([1,2,3]) 313 Xiter([1, 2, 3]).flatten() 314 Xiter([[1, 2], [3]]).flatten() 315 Xiter([[1, 2], 3]).flatten() 316 ``` 317 """ 318 319 def result(xi): 320 for el in xi: 321 for inner_el in el: 322 yield inner_el 323 324 return Xiter(result(self.copy()))
Return a new iterator, with each element nested iterated on individually.
Usage
from xfp import Xiter
# All the following resulting objects are equivalent to Xiter([1,2,3])
Xiter([1, 2, 3]).flatten()
Xiter([[1, 2], [3]]).flatten()
Xiter([[1, 2], 3]).flatten()
def
flat_map(self, f: F1[[+X], typing.Iterable[T]]) -> 'Xiter[T]':
326 def flat_map[T](self, f: F1[[X], Iterable[T]]) -> "Xiter[T]": 327 """Return the result of map and then flatten. 328 329 Exists as homogenisation with Xresult.flat_map. 330 331 ### Usage 332 333 ```python 334 from xfp import Xiter, Xlist 335 336 Xiter([1, 2, 3]).flat_map(lambda x: Xlist([(x, 4), (x, 5)])) 337 # equivalent to Xiter([(1, 4), (1, 5), (2, 4), (2, 5), (3, 4), (3, 5)]) 338 ``` 339 """ 340 return self.map(f).flatten()
Return the result of map and then flatten.
Exists as homogenisation with Xresult.flat_map.
Usage
from xfp import Xiter, Xlist
Xiter([1, 2, 3]).flat_map(lambda x: Xlist([(x, 4), (x, 5)]))
# equivalent to Xiter([(1, 4), (1, 5), (2, 4), (2, 5), (3, 4), (3, 5)])
def
fold_left(self, zero: T, f: F1[[T, +X], T]) -> T:
342 def fold_left[T](self, zero: T, f: F1[[T, X], T]) -> T: 343 """Return the accumulation of the Xiter elements. 344 345 - Uses a custom accumulator (zero, f) to aggregate the elements of the Xiter 346 - Initialize the accumulator with the zero value 347 - Then from the first to the last element, compute accumulator(n+1) using f, accumulator(n) and self.data[n], such as: 348 accumulator(n+1) = f(accumulator(n), self.data[n]) 349 - Return the last state of the accumulator 350 351 ### Keyword Arguments 352 353 - zero -- initial state of the accumulator 354 - f -- accumulation function, compute the next state of the accumulator 355 356 ### Warnings 357 358 This function falls in infinite loop in the case of infinite iterator. 359 360 ### Usage 361 362 ```python 363 from xfp import Xiter 364 365 assert Xiter([1, 2, 3]).fold_left(0)(lambda x, y: x + y) == 6 366 assert Xiter([1, 2, 3]).fold_left(10)(lambda x, y: x + y) == 16 367 assert Xiter(["1", "2", "3"]).fold_left("")(lambda x, y: x + y) == "123" 368 assert Xiter([]).fold_left(0)(lambda x, y: x + y) == 0 369 ``` 370 """ 371 acc: T = zero 372 for e in self: 373 acc = f(acc, e) 374 return acc
Return the accumulation of the Xiter elements.
- Uses a custom accumulator (zero, f) to aggregate the elements of the Xiter
- Initialize the accumulator with the zero value
- Then from the first to the last element, compute accumulator(n+1) using f, accumulator(n) and self.data[n], such as: accumulator(n+1) = f(accumulator(n), self.data[n])
- Return the last state of the accumulator
Keyword Arguments
- zero -- initial state of the accumulator
- f -- accumulation function, compute the next state of the accumulator
Warnings
This function falls in infinite loop in the case of infinite iterator.
Usage
from xfp import Xiter
assert Xiter([1, 2, 3]).fold_left(0)(lambda x, y: x + y) == 6
assert Xiter([1, 2, 3]).fold_left(10)(lambda x, y: x + y) == 16
assert Xiter(["1", "2", "3"]).fold_left("")(lambda x, y: x + y) == "123"
assert Xiter([]).fold_left(0)(lambda x, y: x + y) == 0
def
fold(self, zero: T, f: F1[[T, +X], T]) -> T:
376 def fold[T](self, zero: T, f: F1[[T, X], T]) -> T: 377 """Return the accumulation of the Xiter elements. 378 379 Shorthand for fold_left 380 """ 381 return self.fold_left(zero, f)
Return the accumulation of the Xiter elements.
Shorthand for fold_left
def
reduce(self, f: F1[[+X, +X], +X]) -> +X:
383 def reduce(self, f: F1[[X, X], X]) -> X: 384 """Return the accumulation of the Xiter elements using the first element as the initial state of accumulation. 385 386 ### Raise 387 388 - IndexError -- when the Xiter is empty 389 390 ### Keyword Arguments 391 392 - f -- accumulation function, compute the next state of the accumulator 393 394 ### Warning 395 396 This function falls in infinite loop in the case of infinite iterator. 397 398 ### Usage 399 400 ```python 401 from xfp import Xiter 402 import pytest 403 404 assert Xiter([1, 2, 3]).reduce(lambda x, y: x + y) == 6 405 assert Xiter(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123" 406 with pytest.raises(IndexError): 407 Xiter([]).reduce(lambda x, y: x + y) 408 ``` 409 """ 410 try: 411 h = self.head() 412 except IndexError: 413 raise IndexError("<reduce> operation not allowed on empty list") 414 return self.tail().fold(h, f)
Return the accumulation of the Xiter elements using the first element as the initial state of accumulation.
Raise
- IndexError -- when the Xiter is empty
Keyword Arguments
- f -- accumulation function, compute the next state of the accumulator
Warning
This function falls in infinite loop in the case of infinite iterator.
Usage
from xfp import Xiter
import pytest
assert Xiter([1, 2, 3]).reduce(lambda x, y: x + y) == 6
assert Xiter(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123"
with pytest.raises(IndexError):
Xiter([]).reduce(lambda x, y: x + y)
416 def reduce_fr(self, f: F1[[X, X], X]) -> Xresult[IndexError, X]: 417 """Return the accumulation of the Xiter elements using the first element as the initial state of accumulation. 418 419 Wrap the potential error in an Xresult. 420 421 ### Keyword Arguments 422 423 - f -- accumulation function, compute the next state of the accumulator 424 425 ### Warning 426 427 This function falls in infinite loop in the case of infinite iterator. 428 429 ### Usage 430 431 ```python 432 from xfp import Xiter, Xtry 433 434 Xiter([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6) 435 Xiter(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123") 436 Xiter([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list")) 437 438 ``` 439 """ 440 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.reduce(f)))
Return the accumulation of the Xiter elements using the first element as the initial state of accumulation.
Wrap the potential error in an Xresult.
Keyword Arguments
- f -- accumulation function, compute the next state of the accumulator
Warning
This function falls in infinite loop in the case of infinite iterator.
Usage
from xfp import Xiter, Xtry
Xiter([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6)
Xiter(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123")
Xiter([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list"))
def
min(self, key: F1[[+X], _Comparable] = <built-in function id>) -> +X:
442 def min(self, key: F1[[X], _Comparable] = id) -> X: 443 """Return the smallest element of the Xiter given the key criteria. 444 445 ### Raise 446 447 - ValueError -- when the Xiter is empty 448 449 ### Keyword Arguments 450 451 - key (default id) -- the function which extrapolate a sortable from the elements of the list 452 453 ### Warning 454 455 This function falls in infinite loop in the case of infinite iterator. 456 457 ### Usage 458 459 ```python 460 from xfp import Xiter 461 import pytest 462 463 input = Xiter(["ae", "bd", "cc"]) 464 assert input.min() == "ae" 465 assert input.min(lambda x: x[-1]) == "cc" 466 with pytest.raises(IndexError): 467 Xiter([]).min() 468 ``` 469 """ 470 return min(self, key=key)
Return the smallest element of the Xiter given the key criteria.
Raise
- ValueError -- when the Xiter is empty
Keyword Arguments
- key (default id) -- the function which extrapolate a sortable from the elements of the list
Warning
This function falls in infinite loop in the case of infinite iterator.
Usage
from xfp import Xiter
import pytest
input = Xiter(["ae", "bd", "cc"])
assert input.min() == "ae"
assert input.min(lambda x: x[-1]) == "cc"
with pytest.raises(IndexError):
Xiter([]).min()
472 def min_fr(self, key: F1[[X], _Comparable] = id) -> Xresult[ValueError, X]: 473 """Return the smallest element of the Xiter given the key criteria. 474 475 Wrap the potential failure in an Wresult 476 477 ### Warning 478 479 This function falls in infinite loop in the case of infinite iterator. 480 ``` 481 """ 482 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.min(key)))
Return the smallest element of the Xiter given the key criteria.
Wrap the potential failure in an Wresult
Warning
This function falls in infinite loop in the case of infinite iterator. ```
def
max(self, key: F1[[+X], _Comparable] = <built-in function id>) -> +X:
484 def max(self, key: F1[[X], _Comparable] = id) -> X: 485 """Return the bigget element of the Xiter given the key criteria. 486 487 ### Raise 488 489 - ValueError -- when the Xiter is empty 490 491 ### Keyword Arguments 492 493 - key (default id) -- the function which extrapolate a sortable from the elements of the list 494 495 ### Warning 496 497 This function falls in infinite loop in the case of infinite iterator. 498 499 ### Usage 500 501 ```python 502 from xfp import Xiter 503 import pytest 504 505 input = Xiter(["ae", "bd", "cc"]) 506 assert input.max() == "cc" 507 assert input.max(lambda x: x[-1]) == "ae" 508 with pytest.raises(IndexError): 509 Xiter([]).max() 510 ``` 511 """ 512 return max(self, key=key)
Return the bigget element of the Xiter given the key criteria.
Raise
- ValueError -- when the Xiter is empty
Keyword Arguments
- key (default id) -- the function which extrapolate a sortable from the elements of the list
Warning
This function falls in infinite loop in the case of infinite iterator.
Usage
from xfp import Xiter
import pytest
input = Xiter(["ae", "bd", "cc"])
assert input.max() == "cc"
assert input.max(lambda x: x[-1]) == "ae"
with pytest.raises(IndexError):
Xiter([]).max()
514 def max_fr(self, key: F1[[X], _Comparable] = id) -> Xresult[ValueError, X]: 515 """Return the biggest element of the Xiter given the key criteria. 516 517 Wrap the potential failure in an Wresult 518 519 ### Warning 520 521 This function falls in infinite loop in the case of infinite iterator. 522 ``` 523 """ 524 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.max(key)))
Return the biggest element of the Xiter given the key criteria.
Wrap the potential failure in an Wresult
Warning
This function falls in infinite loop in the case of infinite iterator. ```
526 def take(self, n: int) -> "Xiter[X]": 527 """Return a new iterator limited to the first 'n' elements. 528 Return a copy if the original iterator has less than 'n' elements. 529 Return an empty Xiter if n is negative. 530 531 Do not consume the original iterator. 532 533 ### Usage 534 535 ```python 536 from xfp import Xiter 537 import itertools 538 539 infinite_xiter = Xiter(itertools.repeat(42)) # -> Xiter([42,42,42,...]) 540 until_xiter = infinite_xiter.take(3) # -> Xiter([42,42]) 541 ``` 542 """ 543 544 return Xiter(self.slice(n))
Return a new iterator limited to the first 'n' elements. Return a copy if the original iterator has less than 'n' elements. Return an empty Xiter if n is negative.
Do not consume the original iterator.
Usage
from xfp import Xiter
import itertools
infinite_xiter = Xiter(itertools.repeat(42)) # -> Xiter([42,42,42,...])
until_xiter = infinite_xiter.take(3) # -> Xiter([42,42])
546 def takeuntil(self, predicate: F1[[X], bool]) -> "Xiter[X]": 547 """Return a new iterator that stops yielding elements when predicate = True. 548 549 Do not consume the original iterator. 550 551 Useful to limit an infinite Xiter with a predicate. 552 553 ### Usage 554 555 ```python 556 from xfp import Xiter 557 import itertools 558 559 infinite_xiter = Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...]) 560 until_xiter = infinite_xiter.takeuntil(lambda x: x >=6) # -> Xiter([0,2,4]) 561 ``` 562 """ 563 564 # fixes "error: Cannot use a covariant type variable as a parameter" on lambda x: not predicate(x) 565 @curry2 566 def invert[T](p: F1[[T], bool], x: T) -> bool: 567 return not p(x) 568 569 return self.takewhile(invert(predicate))
Return a new iterator that stops yielding elements when predicate = True.
Do not consume the original iterator.
Useful to limit an infinite Xiter with a predicate.
Usage
from xfp import Xiter
import itertools
infinite_xiter = Xiter(itertools.count(start=0,step=2)) # -> Xiter([0,2,4,6,8,...])
until_xiter = infinite_xiter.takeuntil(lambda x: x >=6) # -> Xiter([0,2,4])
579 def slice(self, *args) -> "Xiter[X]": 580 """Return an new Xiter with selected elements from the Xiter. 581 Works like sequence slicing but does not support negative values 582 for start, stop, or step. 583 584 Do not consume the original iterator. 585 586 If start is zero or None, iteration starts at zero. 587 Otherwise, elements from the iterable are skipped until start is reached. 588 589 If stop is None, iteration continues until the input is exhausted, 590 if at all. Otherwise, it stops at the specified position. 591 592 If step is None, the step defaults to one. 593 Elements are returned consecutively unless step is set higher than 594 one which results in items being skipped. 595 """ 596 __iter_copy = self.copy() 597 598 if len(args) not in (1, 2, 3): 599 raise TypeError( 600 "slice expected from 1 to 3 positional arguments: 'stop' | 'start' 'stop' ['step']" 601 ) 602 603 return Xiter(itertools.islice(__iter_copy, *args))
Return an new Xiter with selected elements from the Xiter. Works like sequence slicing but does not support negative values for start, stop, or step.
Do not consume the original iterator.
If start is zero or None, iteration starts at zero. Otherwise, elements from the iterable are skipped until start is reached.
If stop is None, iteration continues until the input is exhausted, if at all. Otherwise, it stops at the specified position.
If step is None, the step defaults to one. Elements are returned consecutively unless step is set higher than one which results in items being skipped.
def
zip(self, other: Iterable[T]) -> 'Xiter[tuple[X, T]]':
605 def zip[T](self, other: Iterable[T]) -> "Xiter[tuple[X, T]]": 606 """Zip this iterator with another iterable.""" 607 return Xiter(zip(self.copy(), other))
Zip this iterator with another iterable.
class
Xlist(typing.Generic[+X]):
51class Xlist(Generic[X]): 52 """Enhance Lists (eager) with functional behaviors. 53 54 This class provides common behaviors used for declarative programming. 55 56 ### Features 57 58 - Monadic behavior 59 - Descriptive accumulation 60 - List proxies or quality of lifes 61 """ 62 63 def __init__(self, iterable: Iterable[X]) -> None: 64 """Construct an Xlist from an iterable.""" 65 match iterable: 66 case ABCIterable(): 67 self.__data = list(iterable) 68 case _: 69 raise TypeError( 70 f"'{type(iterable).__name__}' not allowed for Xlist constructor" 71 ) 72 73 def __iter__(self) -> Iterator[X]: 74 """Return an iterable over the underlying data.""" 75 return iter(self.__data) 76 77 def __eq__(self, other: object) -> bool: 78 """Return the equality by comparison of inner values (and order).""" 79 match other: 80 case ABCIterable(): 81 return [e for e in self] == [e for e in other] 82 case _: 83 return False 84 85 def __len__(self) -> int: 86 """Return the length of the underlying data.""" 87 return len(self.__data) 88 89 def __repr__(self) -> str: 90 """Return the representation of the underlying data""" 91 return f"Xlist({repr(self.__data)})" 92 93 def __getitem__(self, i: int) -> X: 94 """Alias for get(i). 95 96 Exists to enable [] syntax 97 """ 98 return self.get(i) 99 100 def copy(self) -> Xlist[X]: 101 "Return a shallow copy of itself." 102 return Xlist(copy(self.__data)) 103 104 def deepcopy(self) -> Xlist[X]: 105 "Return a deep copy of itself." 106 return Xlist(deepcopy(self.__data)) 107 108 def get(self, i: int) -> X: 109 """Return the i-th element of the Xlist. 110 111 ### Raise 112 113 - IndexError -- if the list is shorter than i 114 """ 115 if len(self) <= i: 116 raise IndexError( 117 f"<get> operation not allowed on list shorter than index {i} (found {len(self)} elements)." 118 ) 119 return self.__data[i] 120 121 def get_fr(self, i: int) -> Xresult[IndexError, X]: 122 """Return the i-th element of the Xlist. 123 124 Wrap the potential error in an Xresult. 125 """ 126 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(i))) 127 128 def head(self) -> X: 129 """Alias for get(0).""" 130 return self.get(0) 131 132 def head_fr(self) -> Xresult[IndexError, X]: 133 """Alias for get_fr(0).""" 134 return self.get_fr(0) 135 136 def tail(self) -> Xlist[X]: 137 """Return the Xlist except its first element. 138 139 ### Raise 140 141 - IndexError -- if the list is empty. 142 """ 143 if len(self) <= 0: 144 raise IndexError("<tail> operation not allowed on empty list") 145 return Xlist(self.__data[1:]) 146 147 def tail_fr(self) -> Xresult[IndexError, Xlist[X]]: 148 """Return the Xlist except its first element. 149 150 Wrap the potential error in an Xresult. 151 """ 152 return cast(Xresult[IndexError, Xlist[X]], Xtry.from_unsafe(self.tail)) 153 154 def appended[T](self: Xlist[T], el: T) -> Xlist[T]: 155 """Return a new Xlist with el appended at its end.""" 156 newlist = self.copy() 157 newlist.__data.append(el) 158 return newlist 159 160 def prepended[T](self: Xlist[T], el: T) -> Xlist[T]: 161 """Return a new Xlist with el prepended at index 0.""" 162 newlist = self.copy() 163 newlist.__data.insert(0, el) 164 return newlist 165 166 def inserted[T](self: Xlist[T], i: int, el: T) -> Xlist[T]: 167 """Return a new Xlist with el inserted before position i.""" 168 newlist = self.copy() 169 newlist.__data.insert(i, el) 170 return newlist 171 172 def map[T](self, f: F1[[X], T]) -> Xlist[T]: 173 """Return a new Xlist with the function f applied to each element. 174 175 ### Usage 176 177 ```python 178 from xfp import Xlist 179 180 input = Xlist([1, 2, 3]) 181 f = lambda el: el*el 182 assert input.map(f) == Xlist([f(1), f(2), f(3)]) # == Xlist([1, 4, 9]) 183 ``` 184 """ 185 return Xlist([f(el) for el in self]) 186 187 def filter(self, predicate: F1[[X], bool]) -> Xlist[X]: 188 """Return a new Xlist containing only the elements for which predicate is True. 189 190 ### Usage 191 192 ```python 193 from xfp import Xlist 194 195 input = Xlist([1, 2, 3, 4]) 196 predicate = lambda el: el % 2 == 0 197 assert input.filter(predicate) == Xlist([2, 4]) # keep only even numbers 198 ``` 199 """ 200 return Xlist([el for el in self if predicate(el)]) 201 202 def foreach(self, statement: F1[[X], Any]) -> None: 203 """Do the 'statement' procedure once for each element of the Xlist. 204 205 ### Usage 206 207 ```python 208 from xfp import Xlist 209 210 input = Xlist([1, 2, 3]) 211 statement = lambda el: print(f"This is an element of the list : {el}") 212 input.foreach(statement) 213 # This is an element of the list : 1 214 # This is an element of the list : 2 215 # This is an element of the list : 3 216 ``` 217 """ 218 [statement(e) for e in self] 219 220 def flatten[XS](self: Xlist[Iterable[XS]]) -> Xlist[XS]: 221 """Return a new Xlist with one less level of nest. 222 223 ### Usage 224 225 ```python 226 from xfp import Xlist 227 228 assert Xlist([1, 2, 3]).flatten() == Xlist([1, 2, 3]) 229 assert Xlist([[1, 2], [3]]).flatten() == Xlist([1, 2, 3]) 230 ``` 231 """ 232 return Xlist([el for els in self for el in els]) 233 234 def flat_map[T](self, f: F1[[X], Iterable[T]]) -> "Xlist[T]": 235 """Return the result of map and then flatten. 236 237 Exists as homogenisation with Xresult.flat_map 238 239 ### Usage 240 241 ```python 242 from xfp import Xlist 243 244 actual = Xlist([1, 2, 3]).flat_map(lambda x: Xlist([x, 5])) 245 expected = Xlist([1, 5, 2, 5, 3, 5]) 246 assert actual == expected 247 ``` 248 """ 249 return self.map(f).flatten() 250 251 @overload 252 def min(self: Xlist[_Comparable]) -> X: 253 """Return the smallest element of the Xlist. Elements must be comparables. 254 255 ### Raise 256 257 - ValueError -- when the Xlist is empty 258 259 ### Usage 260 261 ```python 262 from xfp import Xlist 263 264 input = Xlist(["ae", "bd", "cc"]) 265 assert input.min() == "ae" 266 ``` 267 """ 268 ... 269 270 @overload 271 def min(self, key: F1[[X], _Comparable]) -> X: 272 """Return the smallest element of the Xlist given the key criteria. 273 274 ### Argument 275 276 - key -- the function which extrapolate a sortable from the elements of the list 277 278 ### Raise 279 280 - ValueError -- when the Xlist is empty 281 282 ### Usage 283 284 ```python 285 from xfp import Xlist 286 287 input = Xlist(["ae", "bd", "cc"]) 288 assert input.min(lambda x: x[-1]) == "cc" 289 ``` 290 """ 291 ... 292 293 def min(self, key: Any = None) -> X: 294 return min(self, key=key) 295 296 @overload 297 def min_fr(self: Xlist[_Comparable]) -> Xresult[ValueError, X]: 298 """Return the smallest element of the Xlist. Elements must be comparables. 299 300 Wrap the potential failure in an Xresult. 301 302 ### Usage 303 304 ```python 305 from xfp import Xlist, Xresult, XRBranch 306 307 input = Xlist(["ae", "bd", "cc"]) 308 assert input.min_fr() == Xresult("ae", XRBranch.RIGHT) 309 ``` 310 """ 311 ... 312 313 @overload 314 def min_fr(self, key: F1[[X], _Comparable]) -> Xresult[ValueError, X]: 315 """Return the smallest element of the Xlist given the key criteria. 316 317 Wrap the potential failure in an Xresult. 318 319 ### Argument 320 321 - key -- the function which extrapolate a sortable from the elements of the list 322 323 ### Usage 324 325 ```python 326 from xfp import Xlist 327 328 input = Xlist(["ae", "bd", "cc"]) 329 assert input.min_fr(lambda x: x[-1]) == Xresult("cc", XRBranch.RIGHT) 330 ``` 331 """ 332 ... 333 334 def min_fr[X](self: Xlist[X], key=None) -> Xresult[ValueError, X]: 335 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.min(key))) 336 337 @overload 338 def max(self: Xlist[_Comparable]) -> X: 339 """Return the biggest element of the Xlist. Elements must be comparables. 340 341 ### Raise 342 343 - ValueError -- when the Xlist is empty 344 345 ### Usage 346 347 ```python 348 from xfp import Xlist 349 350 input = Xlist(["ae", "bd", "cc"]) 351 assert input.max() == "cc" 352 ``` 353 """ 354 ... 355 356 @overload 357 def max(self, key: F1[[X], _Comparable]) -> X: 358 """Return the biggest element of the Xlist given the key criteria. 359 360 ### Argument 361 362 - key -- the function which extrapolate a sortable from the elements of the list 363 364 ### Raise 365 366 - ValueError -- when the Xlist is empty 367 368 ### Usage 369 370 ```python 371 from xfp import Xlist 372 373 input = Xlist(["ae", "bd", "cc"]) 374 assert input.max(lambda x: x[-1]) == "ae" 375 ``` 376 """ 377 ... 378 379 def max(self, key: Any = None) -> X: 380 return max(self, key=key) 381 382 @overload 383 def max_fr(self: Xlist[_Comparable]) -> Xresult[ValueError, X]: 384 """Return the biggest element of the Xlist. Elements must be comparables. 385 386 Wrap the potential failure in an Xresult. 387 388 ### Usage 389 390 ```python 391 from xfp import Xlist, XResult, XRBranch 392 393 input = Xlist(["ae", "bd", "cc"]) 394 assert input.max_fr() == Xresult("cc", XRBranch.RIGHT) 395 ``` 396 """ 397 ... 398 399 @overload 400 def max_fr(self, key: F1[[X], _Comparable]) -> Xresult[ValueError, X]: 401 """Return the biggest element of the Xlist given the key criteria. 402 403 Wrap the potential failure in an Wresult. 404 405 ### Argument 406 407 - key -- the function which extrapolate a sortable from the elements of the list 408 409 ### Usage 410 411 ```python 412 from xfp import Xlist, XResult, XRBranch 413 414 input = Xlist(["ae", "bd", "cc"]) 415 assert input.max_fr(lambda x: x[-1]) == Xresult("ae", XRBranch.RIGHT) 416 ``` 417 """ 418 ... 419 420 def max_fr[X](self: Xlist[X], key=None) -> Xresult[ValueError, X]: 421 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.max(key))) 422 423 @overload 424 def sorted(self: Xlist[_Comparable], *, reverse: bool = False) -> Xlist[X]: 425 """Return a new Xlist containing the same elements sorted. Elements must be comparables. 426 427 ### Keyword Arguments 428 429 - reverse (default False) -- should we sort ascending (False) or descending (True) 430 431 ### Usage 432 433 ```python 434 from xfp import Xlist 435 436 input = Xlist(["bd", "ae", "cc"]) 437 assert input.sorted() == Xlist(["ae", "bd", "cc"]) 438 assert input.sorted(reverse=True) == Xlist(["cc", "bd", "ae"]) 439 440 ``` 441 """ 442 ... 443 444 @overload 445 def sorted(self, *, key: F1[[X], _Comparable], reverse: bool = False) -> Xlist[X]: 446 """Return a new Xlist containing the same elements sorted given the key criteria. 447 448 ### Keyword Arguments 449 450 - key -- the function which extrapolate a sortable from the elements of the list 451 - reverse (default False) -- should we sort ascending (False) or descending (True) 452 453 ### Usage 454 455 ```python 456 from xfp import Xlist 457 458 input = Xlist(["bd", "ae", "cc"]) 459 assert input.sorted() == Xlist(["ae", "bd", "cc"]) 460 assert input.sorted(lambda x: x[-1]) == Xlist(["cc", "bd", "ae"]) 461 assert input.sorted(lambda x: x[-1], reverse = True) == Xlist(["ae", "bd", "cc"]) 462 ``` 463 """ 464 465 def sorted(self, key: Any = None, reverse: bool = False) -> Xlist[X]: 466 return Xlist(sorted(self, key=key, reverse=reverse)) 467 468 def reversed(self) -> Xlist[X]: 469 """Return a new Xlist containing the same elements in the reverse order.""" 470 data: list[X] = self.__data.copy() 471 data.reverse() 472 return Xlist(data) 473 474 @overload 475 def fold_left[Y](self, zero: Y, f: F1[[Y, X], Y]) -> Y: 476 """Return the accumulation of the Xlist elements. 477 478 - Uses a custom accumulator (zero, f) to aggregate the elements of the Xlist 479 - Initialize the accumulator with the zero value 480 - Then from the first to the last element, compute accumulator(n+1) using f, accumulator(n) and self.data[n], such as: 481 accumulator(n+1) = f(accumulator(n), self.data[n]) 482 - Return the last state of the accumulator 483 484 ### Arguments 485 486 - zero -- initial state of the accumulator 487 - f -- accumulation function, compute the next state of the accumulator 488 489 ### Usage 490 491 ```python 492 from xfp import Xlist 493 494 assert Xlist([1, 2, 3]).fold_left(0, lambda x, y: x + y) == 6 495 assert Xlist([1, 2, 3]).fold_left(10, lambda x, y: x + y) == 16 496 assert Xlist(["1", "2", "3"]).fold_left("", lambda x, y: x + y) == "123" 497 assert Xlist([]).fold_left(0, lambda x, y: x + y) == 0 498 ``` 499 """ 500 ... 501 502 @overload 503 def fold_left[Y](self, zero: Y) -> F1[[F1[[Y, X], Y]], Y]: 504 """@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. 505 Return the accumulation of the Xlist elements - Curried Version.""" 506 ... 507 508 def fold_left[Y]( 509 self, zero: Y, f: F1[[Y, X], Y] | None = None 510 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 511 def _fold_left(zero: Y, f: F1[[Y, X], Y]) -> Y: 512 acc = zero 513 for e in self: 514 acc = f(acc, e) 515 return acc 516 517 if f is None: 518 return _partially_apply_fold(_fold_left, zero) 519 else: 520 return _fold_left(zero, f) 521 522 @overload 523 def fold_right[Y](self, zero: Y, f: F1[[Y, X], Y]) -> Y: 524 """Return the accumulation of the Xlist elements. 525 526 - Uses a custom accumulator (zero, f) to aggregate the elements of the Xlist 527 - Initialize the accumulator with the zero value 528 - Then from the last to the first element, compute accumulator(n+1) using f, accumulator(n) and self.data[n], such as: 529 accumulator(n+1) = f(accumulator(n), self.data[n]) 530 - Return the last state of the accumulator 531 532 ### Keyword Arguments 533 534 - zero -- initial state of the accumulator 535 - f -- accumulation function, compute the next state of the accumulator 536 537 ### Usage 538 539 ```python 540 from xfp import Xlist 541 542 assert Xlist([1, 2, 3]).fold_right(0, lambda x, y: x + y) == 6 543 assert Xlist([1, 2, 3]).fold_right(10, lambda x, y: x + y) == 16 544 assert Xlist(["1", "2", "3"]).fold_right("", lambda x, y: x + y) == "321" 545 assert Xlist([]).fold_right(0, lambda x, y: x + y) == 0 546 ``` 547 """ 548 ... 549 550 @overload 551 def fold_right[Y](self, zero: Y) -> F1[[F1[[Y, X], Y]], Y]: 552 """@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. 553 Return the accumulation of the Xlist elements - Curried Version.""" 554 ... 555 556 def fold_right[Y]( 557 self, zero: Y, f: F1[[Y, X], Y] | None = None 558 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 559 def _fold_right(zero: Y, f: F1[[Y, X], Y]) -> Y: 560 acc: Y = zero 561 for e in self.reversed(): 562 acc = f(acc, e) 563 return acc 564 565 if f is None: 566 return _partially_apply_fold(_fold_right, zero) 567 else: 568 return _fold_right(zero, f) 569 570 @overload 571 def fold[Y](self, zero: Y, f: F1[[Y, X], Y]) -> Y: 572 """Return the accumulation of the Xlist elements. 573 574 Shorthand for fold_left 575 """ 576 ... 577 578 @overload 579 def fold[Y](self, zero: Y) -> F1[[F1[[Y, X], Y]], Y]: 580 """@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. 581 Return the accumulation of the Xlist elements - Curried Version.""" 582 ... 583 584 def fold[Y]( 585 self, zero: Y, f: F1[[Y, X], Y] | None = None 586 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 587 if f is None: 588 return self.fold_left(zero) 589 else: 590 return self.fold_left(zero, f) 591 592 def reduce(self, f: F1[[X, X], X]) -> X: 593 """Return the accumulation of the Xlist elements using the first element as the initial state of accumulation. 594 595 ### Raise 596 597 - IndexError -- when the Xlist is empty 598 599 ### Keyword Arguments 600 601 - f -- accumulation function, compute the next state of the accumulator 602 603 ### Usage 604 605 ```python 606 from xfp import Xlist 607 import pytest 608 609 assert Xlist([1, 2, 3]).reduce(lambda x, y: x + y) == 6 610 assert Xlist(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123" 611 with pytest.raises(IndexError): 612 Xlist([]).reduce(lambda x, y: x + y) 613 ``` 614 """ 615 if len(self) <= 0: 616 raise IndexError("<reduce> operation not allowed on empty list") 617 return self.tail().fold(self.head())(f) 618 619 def reduce_fr(self, f: F1[[X, X], X]) -> Xresult[IndexError, X]: 620 """Return the accumulation of the Xlist elements using the first element as the initial state of accumulation. 621 622 Wrap the potential error in an Xresult. 623 624 ### Keyword Arguments 625 626 - f -- accumulation function, compute the next state of the accumulator 627 628 ### Usage 629 630 ```python 631 from xfp import Xlist, Xtry 632 633 Xlist([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6) 634 Xlist(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123") 635 Xlist([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list")) 636 637 ``` 638 """ 639 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.reduce(f))) 640 641 def zip[T](self, other: Iterable[T]) -> "Xlist[tuple[X, T]]": 642 """Zip this Xlist with another iterable.""" 643 return Xlist(zip(self, other))
Enhance Lists (eager) with functional behaviors.
This class provides common behaviors used for declarative programming.
Features
- Monadic behavior
- Descriptive accumulation
- List proxies or quality of lifes
Xlist(iterable: Iterable[+X])
63 def __init__(self, iterable: Iterable[X]) -> None: 64 """Construct an Xlist from an iterable.""" 65 match iterable: 66 case ABCIterable(): 67 self.__data = list(iterable) 68 case _: 69 raise TypeError( 70 f"'{type(iterable).__name__}' not allowed for Xlist constructor" 71 )
Construct an Xlist from an iterable.
100 def copy(self) -> Xlist[X]: 101 "Return a shallow copy of itself." 102 return Xlist(copy(self.__data))
Return a shallow copy of itself.
104 def deepcopy(self) -> Xlist[X]: 105 "Return a deep copy of itself." 106 return Xlist(deepcopy(self.__data))
Return a deep copy of itself.
def
get(self, i: int) -> +X:
108 def get(self, i: int) -> X: 109 """Return the i-th element of the Xlist. 110 111 ### Raise 112 113 - IndexError -- if the list is shorter than i 114 """ 115 if len(self) <= i: 116 raise IndexError( 117 f"<get> operation not allowed on list shorter than index {i} (found {len(self)} elements)." 118 ) 119 return self.__data[i]
Return the i-th element of the Xlist.
Raise
- IndexError -- if the list is shorter than i
121 def get_fr(self, i: int) -> Xresult[IndexError, X]: 122 """Return the i-th element of the Xlist. 123 124 Wrap the potential error in an Xresult. 125 """ 126 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(i)))
Return the i-th element of the Xlist.
Wrap the potential error in an Xresult.
132 def head_fr(self) -> Xresult[IndexError, X]: 133 """Alias for get_fr(0).""" 134 return self.get_fr(0)
Alias for get_fr(0).
136 def tail(self) -> Xlist[X]: 137 """Return the Xlist except its first element. 138 139 ### Raise 140 141 - IndexError -- if the list is empty. 142 """ 143 if len(self) <= 0: 144 raise IndexError("<tail> operation not allowed on empty list") 145 return Xlist(self.__data[1:])
Return the Xlist except its first element.
Raise
- IndexError -- if the list is empty.
147 def tail_fr(self) -> Xresult[IndexError, Xlist[X]]: 148 """Return the Xlist except its first element. 149 150 Wrap the potential error in an Xresult. 151 """ 152 return cast(Xresult[IndexError, Xlist[X]], Xtry.from_unsafe(self.tail))
Return the Xlist except its first element.
Wrap the potential error in an Xresult.
def
appended(self: 'Xlist[T]', el: 'T') -> 'Xlist[T]':
154 def appended[T](self: Xlist[T], el: T) -> Xlist[T]: 155 """Return a new Xlist with el appended at its end.""" 156 newlist = self.copy() 157 newlist.__data.append(el) 158 return newlist
Return a new Xlist with el appended at its end.
def
prepended(self: 'Xlist[T]', el: 'T') -> 'Xlist[T]':
160 def prepended[T](self: Xlist[T], el: T) -> Xlist[T]: 161 """Return a new Xlist with el prepended at index 0.""" 162 newlist = self.copy() 163 newlist.__data.insert(0, el) 164 return newlist
Return a new Xlist with el prepended at index 0.
def
inserted(self: 'Xlist[T]', i: int, el: 'T') -> 'Xlist[T]':
166 def inserted[T](self: Xlist[T], i: int, el: T) -> Xlist[T]: 167 """Return a new Xlist with el inserted before position i.""" 168 newlist = self.copy() 169 newlist.__data.insert(i, el) 170 return newlist
Return a new Xlist with el inserted before position i.
def
map(self, f: 'F1[[X], T]') -> 'Xlist[T]':
172 def map[T](self, f: F1[[X], T]) -> Xlist[T]: 173 """Return a new Xlist with the function f applied to each element. 174 175 ### Usage 176 177 ```python 178 from xfp import Xlist 179 180 input = Xlist([1, 2, 3]) 181 f = lambda el: el*el 182 assert input.map(f) == Xlist([f(1), f(2), f(3)]) # == Xlist([1, 4, 9]) 183 ``` 184 """ 185 return Xlist([f(el) for el in self])
Return a new Xlist with the function f applied to each element.
Usage
from xfp import Xlist
input = Xlist([1, 2, 3])
f = lambda el: el*el
assert input.map(f) == Xlist([f(1), f(2), f(3)]) # == Xlist([1, 4, 9])
187 def filter(self, predicate: F1[[X], bool]) -> Xlist[X]: 188 """Return a new Xlist containing only the elements for which predicate is True. 189 190 ### Usage 191 192 ```python 193 from xfp import Xlist 194 195 input = Xlist([1, 2, 3, 4]) 196 predicate = lambda el: el % 2 == 0 197 assert input.filter(predicate) == Xlist([2, 4]) # keep only even numbers 198 ``` 199 """ 200 return Xlist([el for el in self if predicate(el)])
Return a new Xlist containing only the elements for which predicate is True.
Usage
from xfp import Xlist
input = Xlist([1, 2, 3, 4])
predicate = lambda el: el % 2 == 0
assert input.filter(predicate) == Xlist([2, 4]) # keep only even numbers
def
foreach(self, statement: F1[[+X], typing.Any]) -> None:
202 def foreach(self, statement: F1[[X], Any]) -> None: 203 """Do the 'statement' procedure once for each element of the Xlist. 204 205 ### Usage 206 207 ```python 208 from xfp import Xlist 209 210 input = Xlist([1, 2, 3]) 211 statement = lambda el: print(f"This is an element of the list : {el}") 212 input.foreach(statement) 213 # This is an element of the list : 1 214 # This is an element of the list : 2 215 # This is an element of the list : 3 216 ``` 217 """ 218 [statement(e) for e in self]
Do the 'statement' procedure once for each element of the Xlist.
Usage
from xfp import Xlist
input = Xlist([1, 2, 3])
statement = lambda el: print(f"This is an element of the list : {el}")
input.foreach(statement)
# This is an element of the list : 1
# This is an element of the list : 2
# This is an element of the list : 3
def
flatten(self: 'Xlist[Iterable[XS]]') -> 'Xlist[XS]':
220 def flatten[XS](self: Xlist[Iterable[XS]]) -> Xlist[XS]: 221 """Return a new Xlist with one less level of nest. 222 223 ### Usage 224 225 ```python 226 from xfp import Xlist 227 228 assert Xlist([1, 2, 3]).flatten() == Xlist([1, 2, 3]) 229 assert Xlist([[1, 2], [3]]).flatten() == Xlist([1, 2, 3]) 230 ``` 231 """ 232 return Xlist([el for els in self for el in els])
Return a new Xlist with one less level of nest.
Usage
from xfp import Xlist
assert Xlist([1, 2, 3]).flatten() == Xlist([1, 2, 3])
assert Xlist([[1, 2], [3]]).flatten() == Xlist([1, 2, 3])
def
flat_map(self, f: 'F1[[X], Iterable[T]]') -> "'Xlist[T]'":
234 def flat_map[T](self, f: F1[[X], Iterable[T]]) -> "Xlist[T]": 235 """Return the result of map and then flatten. 236 237 Exists as homogenisation with Xresult.flat_map 238 239 ### Usage 240 241 ```python 242 from xfp import Xlist 243 244 actual = Xlist([1, 2, 3]).flat_map(lambda x: Xlist([x, 5])) 245 expected = Xlist([1, 5, 2, 5, 3, 5]) 246 assert actual == expected 247 ``` 248 """ 249 return self.map(f).flatten()
Return the result of map and then flatten.
Exists as homogenisation with Xresult.flat_map
Usage
from xfp import Xlist
actual = Xlist([1, 2, 3]).flat_map(lambda x: Xlist([x, 5]))
expected = Xlist([1, 5, 2, 5, 3, 5])
assert actual == expected
def
min(self, key: Any = None) -> +X:
Return the smallest element of the Xlist given the key criteria.
Argument
- key -- the function which extrapolate a sortable from the elements of the list
Raise
- ValueError -- when the Xlist is empty
Usage
from xfp import Xlist
input = Xlist(["ae", "bd", "cc"])
assert input.min(lambda x: x[-1]) == "cc"
334 def min_fr[X](self: Xlist[X], key=None) -> Xresult[ValueError, X]: 335 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.min(key)))
Return the smallest element of the Xlist given the key criteria.
Wrap the potential failure in an Xresult.
Argument
- key -- the function which extrapolate a sortable from the elements of the list
Usage
from xfp import Xlist
input = Xlist(["ae", "bd", "cc"])
assert input.min_fr(lambda x: x[-1]) == Xresult("cc", XRBranch.RIGHT)
def
max(self, key: Any = None) -> +X:
Return the biggest element of the Xlist given the key criteria.
Argument
- key -- the function which extrapolate a sortable from the elements of the list
Raise
- ValueError -- when the Xlist is empty
Usage
from xfp import Xlist
input = Xlist(["ae", "bd", "cc"])
assert input.max(lambda x: x[-1]) == "ae"
420 def max_fr[X](self: Xlist[X], key=None) -> Xresult[ValueError, X]: 421 return cast(Xresult[ValueError, X], Xtry.from_unsafe(lambda: self.max(key)))
Return the biggest element of the Xlist given the key criteria.
Wrap the potential failure in an Wresult.
Argument
- key -- the function which extrapolate a sortable from the elements of the list
Usage
from xfp import Xlist, XResult, XRBranch
input = Xlist(["ae", "bd", "cc"])
assert input.max_fr(lambda x: x[-1]) == Xresult("ae", XRBranch.RIGHT)
465 def sorted(self, key: Any = None, reverse: bool = False) -> Xlist[X]: 466 return Xlist(sorted(self, key=key, reverse=reverse))
Return a new Xlist containing the same elements sorted given the key criteria.
Keyword Arguments
- key -- the function which extrapolate a sortable from the elements of the list
- reverse (default False) -- should we sort ascending (False) or descending (True)
Usage
from xfp import Xlist
input = Xlist(["bd", "ae", "cc"])
assert input.sorted() == Xlist(["ae", "bd", "cc"])
assert input.sorted(lambda x: x[-1]) == Xlist(["cc", "bd", "ae"])
assert input.sorted(lambda x: x[-1], reverse = True) == Xlist(["ae", "bd", "cc"])
468 def reversed(self) -> Xlist[X]: 469 """Return a new Xlist containing the same elements in the reverse order.""" 470 data: list[X] = self.__data.copy() 471 data.reverse() 472 return Xlist(data)
Return a new Xlist containing the same elements in the reverse order.
def
fold_left( self, zero: 'Y', f: 'F1[[Y, X], Y] | None' = None) -> 'F1[[F1[[Y, X], Y]], Y] | Y':
508 def fold_left[Y]( 509 self, zero: Y, f: F1[[Y, X], Y] | None = None 510 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 511 def _fold_left(zero: Y, f: F1[[Y, X], Y]) -> Y: 512 acc = zero 513 for e in self: 514 acc = f(acc, e) 515 return acc 516 517 if f is None: 518 return _partially_apply_fold(_fold_left, zero) 519 else: 520 return _fold_left(zero, f)
@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. Return the accumulation of the Xlist elements - Curried Version.
def
fold_right( self, zero: 'Y', f: 'F1[[Y, X], Y] | None' = None) -> 'F1[[F1[[Y, X], Y]], Y] | Y':
556 def fold_right[Y]( 557 self, zero: Y, f: F1[[Y, X], Y] | None = None 558 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 559 def _fold_right(zero: Y, f: F1[[Y, X], Y]) -> Y: 560 acc: Y = zero 561 for e in self.reversed(): 562 acc = f(acc, e) 563 return acc 564 565 if f is None: 566 return _partially_apply_fold(_fold_right, zero) 567 else: 568 return _fold_right(zero, f)
@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. Return the accumulation of the Xlist elements - Curried Version.
def
fold( self, zero: 'Y', f: 'F1[[Y, X], Y] | None' = None) -> 'F1[[F1[[Y, X], Y]], Y] | Y':
584 def fold[Y]( 585 self, zero: Y, f: F1[[Y, X], Y] | None = None 586 ) -> F1[[F1[[Y, X], Y]], Y] | Y: 587 if f is None: 588 return self.fold_left(zero) 589 else: 590 return self.fold_left(zero, f)
@deprecated since 1.1.0 - Use method with both 'zero' and 'f' arguments instead. Return the accumulation of the Xlist elements - Curried Version.
def
reduce(self, f: F1[[+X, +X], +X]) -> +X:
592 def reduce(self, f: F1[[X, X], X]) -> X: 593 """Return the accumulation of the Xlist elements using the first element as the initial state of accumulation. 594 595 ### Raise 596 597 - IndexError -- when the Xlist is empty 598 599 ### Keyword Arguments 600 601 - f -- accumulation function, compute the next state of the accumulator 602 603 ### Usage 604 605 ```python 606 from xfp import Xlist 607 import pytest 608 609 assert Xlist([1, 2, 3]).reduce(lambda x, y: x + y) == 6 610 assert Xlist(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123" 611 with pytest.raises(IndexError): 612 Xlist([]).reduce(lambda x, y: x + y) 613 ``` 614 """ 615 if len(self) <= 0: 616 raise IndexError("<reduce> operation not allowed on empty list") 617 return self.tail().fold(self.head())(f)
Return the accumulation of the Xlist elements using the first element as the initial state of accumulation.
Raise
- IndexError -- when the Xlist is empty
Keyword Arguments
- f -- accumulation function, compute the next state of the accumulator
Usage
from xfp import Xlist
import pytest
assert Xlist([1, 2, 3]).reduce(lambda x, y: x + y) == 6
assert Xlist(["1", "2", "3"]).reduce(lambda x, y: x + y) == "123"
with pytest.raises(IndexError):
Xlist([]).reduce(lambda x, y: x + y)
619 def reduce_fr(self, f: F1[[X, X], X]) -> Xresult[IndexError, X]: 620 """Return the accumulation of the Xlist elements using the first element as the initial state of accumulation. 621 622 Wrap the potential error in an Xresult. 623 624 ### Keyword Arguments 625 626 - f -- accumulation function, compute the next state of the accumulator 627 628 ### Usage 629 630 ```python 631 from xfp import Xlist, Xtry 632 633 Xlist([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6) 634 Xlist(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123") 635 Xlist([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list")) 636 637 ``` 638 """ 639 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.reduce(f)))
Return the accumulation of the Xlist elements using the first element as the initial state of accumulation.
Wrap the potential error in an Xresult.
Keyword Arguments
- f -- accumulation function, compute the next state of the accumulator
Usage
from xfp import Xlist, Xtry
Xlist([1, 2, 3]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success(6)
Xlist(["1", "2", "3"]).reduce_fr(lambda x, y: x + y) # -> Xtry.Success("123")
Xlist([]).reduce_fr(lambda x, y: x + y) # -> Xtry.Failure(IndexError("<reduce> operation not allowed on empty list"))
@dataclass(frozen=True)
class
Xresult25@dataclass(frozen=True) 26class Xresult[Y, X]: 27 """Encapsulate Union type in container. 28 29 Semantically, Xresult helps managing unpure types, such as : 30 - nullable values (Xresult[None, X]) 31 - tryable values (Xresult[Error, X]) 32 33 Eventually using it for plain encapsulated Union type is fine. 34 35 ### Attributes 36 37 - branch -- XRBranch (LEFT, RIGHT), semaphore telling if the value attribute is an X (LEFT) or a Y (RIGHT) 38 - value -- Y | X, the content of the container 39 40 ### Features 41 42 - Monadic behavior 43 - Manual handling of the bias 44 - Class method to lift values to Xresult 45 46 ### Helper classes 47 48 Helper classes are provided to semantically instantiate / pattern match your Xresults. 49 See Xeither, Xtry, Xopt for more information. 50 51 ### Usages 52 - directly returns Xresult in your own functions: 53 54 ```python 55 def f_that_breaks(should_break: bool) -> Xresult[Exception, str]: 56 if should_break: 57 return Xresult(Exception("something went wrong"), XRBranch.LEFT) 58 else: 59 return Xresult("Everything's fine", XRBranch.RIGHT) 60 ``` 61 62 - catch common functions into Xresult: 63 64 ```python 65 effect_result: Xresult[Exception, E] = Xtry.from_unsafe(some_function_that_raises) 66 ``` 67 68 - powerful optional handling: 69 70 ```python 71 optional_value: int | None = 3 72 option_value: Xresult[None, int] = Xopt.from_optional(optional_value) 73 ``` 74 75 - rich union type: 76 77 ```python 78 def returns_a_str_or_int(should_be_str: boolean) -> Xresult[str, int]: 79 if should_be_str: 80 return Xresult("foo", XRBranch.LEFT) 81 else: 82 return Xresult(42, XRBranch.RIGHT) 83 ``` 84 """ 85 86 value: Y | X 87 branch: XRBranch 88 89 def __eq__(self, value: object) -> bool: 90 return ( 91 isinstance(value, Xresult) 92 and self.value == value.value 93 and self.branch == value.branch 94 ) 95 96 def __repr__(self) -> str: 97 return f"{self.branch} : {self.value}" 98 99 def __iter__(self) -> Iterator[X]: 100 """Return a tri-state iterator of this Xresult. 101 102 Exists as a syntax enhancer in co-usage with `Xresult.fors`. 103 No usage of this iterator should be done outside of it since empty iterator will raise instead of stop. 104 Used to compose right pathes together. 105 106 ### Next values 107 108 - if branch = LEFT : raise an Exception wrapping the Xresult to be caught in `Xresult.fors` 109 - if branch = RIGHT : return self.value 110 - if branch = RIGHT (next already called) : raise StopIteration 111 """ 112 113 class Internal(Iterator[X]): 114 def __init__(self): 115 self.called = False 116 117 def __next__(selff) -> X: 118 if selff.called: 119 raise StopIteration 120 if self.branch == XRBranch.RIGHT: 121 selff.called = True 122 return cast(X, self.value) 123 raise XresultError(self) 124 125 return Internal() 126 127 @staticmethod 128 def fors[T](els: "F0[list[T]]") -> "Xresult[XresultError, T]": 129 """Return the Xresult computed in a list comprehension of zipped Xresult. 130 131 Used as a complement of __iter__ to compose multiple results together. 132 133 ### Keyword Arguments 134 135 - els: lazy list of zero or one element. To be mecanically useful, should be computed as a list comprehension. 136 137 ### Usage 138 139 ```python 140 # Return Xresult(6, XRBranch.RIGHT) 141 Xresult.fors(lambda: # lambda to make the computation lazy 142 [ 143 x + y + z # put here your algorithm for composing effect results 144 for x, y, z 145 in zip( # Chain your effects results in a zip 146 Xeither.Right(1), 147 Xeither.Right(2), 148 Xeither.Right(3) 149 ) 150 ] 151 ) 152 153 # If at least one XResult has a different branch than requested, stop at the first encountered 154 # Return Xresult(2, XRBranch.LEFT) 155 Xresult.fors(lambda: 156 [ 157 x + y + z 158 for x, y, z 159 in zip( 160 Xeither.Right(1), 161 Xeither.Left(2), 162 Xeither.Right(3) 163 ) 164 ] 165 ) 166 ``` 167 """ 168 try: 169 if isinstance(value := els()[0], Xresult): 170 return value 171 else: 172 return Xresult(value, XRBranch.RIGHT) 173 except XresultError as e: 174 return Xresult(e, XRBranch.LEFT) 175 176 def map[T](self, f: F1[[X], T]) -> "Xresult[Y, T]": 177 """Alias for map_right.""" 178 return self.map_right(f) 179 180 def map_left[U](self, f: F1[[Y], U]) -> "Xresult[U, X]": 181 """Return either itself or a new Xresult (LEFT) containing the result of f. 182 183 Is mainly used to chain effect free operations. 184 185 ### Return 186 187 - if self is a RIGHT -- self 188 - if self is a LEFT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value) 189 190 ### Usage 191 192 see map_right 193 """ 194 match self: 195 case Xresult(value, XRBranch.LEFT): 196 return Xresult[U, Never](f(cast(Y, value)), XRBranch.LEFT) 197 case _: 198 return cast(Xresult[Never, X], self) 199 200 def map_right[T](self, f: F1[[X], T]) -> "Xresult[Y, T]": 201 """Return either itself or a new Xresult (RIGHT) containing the result of f. 202 203 ### Return 204 205 - if self is a LEFT -- self 206 - if self is a RIGHT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value) 207 208 ### Usage 209 210 ```python 211 def add_three(i: float) -> float: 212 return i + 3 213 214 def pow(i: float) -> float: 215 return i * i 216 217 ( 218 Xopt 219 .from_optional(3) # Xresult(3, XRBranch.RIGHT) 220 .map_right(add_three) # Xresult(6, XRBranch.RIGHT) 221 .map_right(pow) # Xresult(36, XRBranch.RIGHT) 222 .map_right(lambda x: x - 4) # Xresult(32, XRBranch.RIGHT) 223 ) 224 ``` 225 """ 226 match self: 227 case Xresult(value, XRBranch.RIGHT): 228 return Xresult[Never, T](f(cast(X, value)), XRBranch.RIGHT) 229 case _: 230 return cast(Xresult[Y, Never], self) 231 232 def flatten[YS, U, XS]( 233 self: "Xresult[YS, Xresult[U, XS]]", 234 ) -> "Xresult[YS, Never] | Xresult[U, XS]": 235 """Alias for flatten_right.""" 236 return self.flatten_right() 237 238 def flatten_left[YS, T, XS]( 239 self: "Xresult[Xresult[YS, T], XS]", 240 ) -> "Xresult[YS, T] | Xresult[Never, XS]": 241 """Return either self or a new flat Xresult if the underlying value is an Xresult. 242 243 ### Return 244 245 - if self.value is an Xresult, and branch = LEFT -- a new Xresult being the underlying value 246 - otherwise -- self 247 248 ### Usage 249 250 see flatten_right 251 """ 252 match self: 253 case Xresult(value, XRBranch.LEFT): 254 return cast(Xresult[YS, T], value) 255 case _: 256 return cast(Xresult[Never, XS], self) 257 258 def flatten_right[YS, U, XS]( 259 self: "Xresult[YS, Xresult[U, XS]]", 260 ) -> "Xresult[YS, Never] | Xresult[U, XS]": 261 """Return either self or a new flat Xresult if the underlying value is an Xresult. 262 263 ### Return 264 265 - if self.value is an Xresult, and branch = RIGHT -- a new Xresult being the underlying value 266 - otherwise -- self 267 268 ### Usage 269 270 ```python 271 assert Xresult.right(Xresult.right("example")).flatten_right() == Xresult("example", XRBranch.RIGHT) 272 assert Xresult.right("example").flatten_right() == Xresult("example", XRBranch.RIGHT) 273 assert Xresult.right(Xopt.from_optional(None)).flatten_right() == Xresult(None, XRBranch.LEFT) 274 assert Xresult.right(Xresult.left("example")).flatten_right() == Xresult("example", XRBranch.LEFT) 275 ``` 276 """ 277 match self: 278 case Xresult(value, XRBranch.RIGHT): 279 return cast(Xresult[U, XS], value) 280 case _: 281 return cast(Xresult[YS, Never], self) 282 283 def flat_map[T, U]( 284 self, f: "F1[[X], Xresult[U, T]]" 285 ) -> "Xresult[Y, Never] | Xresult[U, T]": 286 """Alias for flat_map_right.""" 287 return self.flat_map_right(f) 288 289 def flat_map_left[T, U]( 290 self, f: "F1[[Y], Xresult[U, T]]" 291 ) -> "Xresult[U, T] | Xresult[Never, X]": 292 """Return the result of map_left then flatten. 293 294 ### Return 295 296 - if self is a RIGHT -- self 297 - if self is a LEFT -- a new Xresult, map_left then flatten 298 299 ### Usage 300 301 see flat_map 302 """ 303 return self.map_left(f).flatten_left() 304 305 def flat_map_right[T, U]( 306 self, f: "F1[[X], Xresult[U, T]]" 307 ) -> "Xresult[Y, Never] | Xresult[U, T]": 308 """Return the result of map_right then flatten. 309 310 ### Return 311 312 - if self is a LEFT -- self 313 - if self is a RIGHT -- a new Xresult, map_right then flatten 314 315 ### Usage 316 317 ```python 318 from xfp import Xtry 319 import math 320 321 @Xtry.safed 322 def invert(i: float) -> float: 323 return 1 / i 324 325 @Xtry.safed 326 def sqrt(i: float) -> float: 327 return math.sqrt(i) 328 329 ( 330 Xtry 331 .Success(4) # Xresult(4, XRBranch.RIGHT) 332 .flat_map_right(invert) # Xresult(0.25, XRBranch.RIGHT) 333 .flat_map_right(sqrt) # Xresult(0.5, XRBranch.RIGHT) 334 ) 335 ( 336 Xtry 337 .Success(0) # Xresult(0, XRBranch.RIGHT) 338 .flat_map_right(invert) # Xresult(ZeroDivisionError(..., XRBranch.LEFT)) 339 .flat_map_right(sqrt) # Xresult(ZeroDivisionError(..., XRBranch.LEFT)) 340 ) 341 ( 342 Xtry 343 .Success(-4) # Xresult(-4, XRBranch.RIGHT) 344 .flat_map_right(invert) # Xresult(-0.25, XRBranch.RIGHT) 345 .flat_map_right(sqrt) # Xresult(ValueError(..., XRBranch.LEFT)) 346 ) 347 """ 348 return self.map_right(f).flatten_right() 349 350 @curry_method2 351 def fold[T, TT](self, default: TT, f: F1[[X], T]) -> T | TT: 352 """Return default if branch != RIGHT, otherwise f(self.value). 353 354 Exists as homogenisation with Xlist.fold 355 356 ### Keyword Arguments 357 358 - default -- output when the value does not exist on the RIGHT side 359 - f -- transformation to apply to the underlying value before returning 360 when the value is present on the RIGHT side 361 362 ### Usage 363 364 ```python 365 from xfp import Xtry 366 from datetime import date 367 368 def load_dated_partition(partition_value: date) -> str: 369 return "actual partition" 370 371 372 @Xtry.safed 373 def to_date(str_date: str) -> date: 374 print(date.fromisoformat(str_date)) 375 return date.fromisoformat(str_date) 376 377 data = to_date("2024-05-01").fold("default partition")(load_dated_partition) 378 # good date -> date parsed by to_date() -> fold() calls load_dated_partition 379 380 data = to_date("2023-02-29").fold("default partition")(load_dated_partition) 381 # wrong date -> LEFT bias returned by to_date() -> fold returns "default partition" 382 ``` 383 """ 384 if self.branch == XRBranch.RIGHT: 385 return f(cast(X, self.value)) 386 else: 387 return default 388 389 def get_or_else[T](self, default: T) -> X | T: 390 """Shorthand for self.fold(default)(id) 391 392 ### Usage 393 ```python 394 from xfp import Xtry 395 from datetime import date 396 397 @Xtry.safed 398 def to_date(str_date: str) -> date: 399 return date.fromisoformat(str_date) 400 401 gotten_date = to_date("2024-05-01").get_or_else(date.today()) 402 else_date = to_date("2023-02-29").get_or_else(date.today()) 403 ``` 404 """ 405 return self.fold(default)(lambda x: x) 406 407 def foreach(self, statement: F1[[X], Any]) -> None: 408 """Alias for foreach_right.""" 409 self.foreach_right(statement) 410 411 def foreach_left(self, statement: F1[[Y], Any]) -> None: 412 """Do the statement procedure to the underlying value if self is a LEFT. 413 414 ### Usage 415 416 see foreach_right 417 """ 418 if self.branch == XRBranch.LEFT: 419 statement(cast(Y, self.value)) 420 421 def foreach_right(self, statement: F1[[X], Any]) -> None: 422 """Do the statement procedure to the underlying value if self is a RIGHT. 423 424 ### Usage 425 ```python 426 from xfp import Xresult,XRBranch,Xopt 427 428 ( 429 Xopt 430 .from_optional(25) 431 .foreach(lambda x: print(f"This is an element of the list : {x}")) 432 ) 433 # This is an element of the list : 25 434 435 ( 436 Xopt 437 .from_optional(None) 438 .foreach(lambda x: print(f"This is the element : {x}")) 439 ) 440 # doesn't output anything 441 442 ( 443 Xresult(42, XRBranch.RIGHT) 444 .foreach(lambda x: print(f"This is the right element : {x}")) 445 ) 446 # This is the right element : 42 447 448 ( 449 Xresult(666, XRBranch.LEFT) 450 .foreach(lambda x: print(f"This is the right element : {x}")) 451 ) 452 # doesn't output anything 453 ``` 454 """ 455 if self.branch == XRBranch.RIGHT: 456 statement(cast(X, self.value)) 457 458 def recover_with[T, U]( 459 self, f: "F1[[Y], Xresult[U, T]]" 460 ) -> "Xresult[U, T] | Xresult[Never, X]": 461 """Alias for recover_with_right.""" 462 return self.recover_with_right(f) 463 464 def recover_with_left[T, U]( 465 self, f: "F1[[X], Xresult[U, T]]" 466 ) -> "Xresult[Y, Never] | Xresult[U, T]": 467 """Return itself, mapped on the right side, flattened on the left side. 468 469 See flat_map_right 470 471 ### Usage 472 473 See recover_with_right 474 """ 475 match self: 476 case Xresult(v, XRBranch.RIGHT): 477 return f(cast(X, v)) 478 case _: 479 return cast(Xresult[Y, Never], self) 480 481 def recover_with_right[T, U]( 482 self, f: "F1[[Y], Xresult[U, T]]" 483 ) -> "Xresult[U, T] | Xresult[Never, X]": 484 """Return itself, mapped on the left side, flattened on the right side. 485 486 See flat_map_left 487 488 ### Usage 489 490 ```python 491 from xfp import Xtry 492 import math 493 494 @Xtry.safed 495 def invert(i: float) -> float: 496 return 1 / i 497 498 @Xtry.safed 499 def sqrt(i: float) -> float: 500 return math.sqrt(i) 501 502 ( 503 sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT) 504 .recover_with_right(lambda _: invert(-4)) # Xresult(-0.25, XRBranch.RIGHT) 505 ) 506 507 ( 508 sqrt(4) # Xresult(2, XRBranch.RIGHT) 509 .recover_with_right(lambda _: invert(-4)) # Xresult(2, XRBranch.RIGHT) 510 ) 511 ``` 512 """ 513 match self: 514 case Xresult(v, XRBranch.LEFT): 515 return f(cast(Y, v)) 516 case _: 517 return cast(Xresult[Never, X], self) 518 519 def recover[T](self, f: F1[[Y], T]) -> "Xresult[Never, X | T]": 520 """Alias for recover_right.""" 521 return self.recover_right(f) 522 523 def recover_left[U](self, f: F1[[X], U]) -> "Xresult[Y | U, Never]": 524 """Return a new Xresult with is always a LEFT. 525 526 Used to convert a RIGHT result into a LEFT using an effectless transformation. 527 Semantically : 528 Used to fallback from a potential success with an effectless operation. 529 This is a fallback that always ends up as a 'failure'. 530 531 ### Return 532 533 - if branch == LEFT -- self 534 - otherwise -- a new Xresult, having branch = LEFT, with an inherent value of f(self.value) 535 536 ### Usage 537 538 See recover_right 539 """ 540 match self: 541 case Xresult(v, XRBranch.RIGHT): 542 return Xresult(f(cast(X, v)), XRBranch.LEFT) 543 case _: 544 return cast(Xresult[Y, Never], self) 545 546 def recover_right[T](self, f: F1[[Y], T]) -> "Xresult[Never, T | X]": 547 """Return a new Xresult with is always a RIGHT. 548 549 Used to convert a LEFT result into a RIGHT using an effectless transformation. 550 Semantically : 551 Used to fallback from a potential failure with an effectless operation. 552 This is a fallback that always ends up as a 'success'. 553 554 ### Return 555 556 - if branch == RIGHT -- self 557 - otherwise -- a new Xresult, having branch = RIGHT, with an inherent value of f(self.value) 558 559 ### Usage 560 561 ```python 562 from xfp import Xtry 563 import math 564 565 @Xtry.safed 566 def sqrt(i: float) -> float: 567 return math.sqrt(i) 568 569 ( 570 sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT) 571 .recover_right(lambda _: 0) # Xresult(0, XRBranch.RIGHT) 572 ) 573 ( 574 sqrt(4) # Xresult(2, XRBranch.RIGHT) 575 .recover_right(lambda _: 0) # Xresult(2.0, XRBranch.RIGHT) 576 ) 577 ``` 578 """ 579 match self: 580 case Xresult(v, XRBranch.LEFT): 581 return Xresult(f(cast(Y, v)), XRBranch.RIGHT) 582 case _: 583 return cast(Xresult[Never, X], self) 584 585 def filter( 586 self, predicate: F1[[X], bool] 587 ) -> "Xresult[Y, Never] | Xresult[XresultError, X]": 588 """Alias for filter_right.""" 589 return self.filter_right(predicate) 590 591 def filter_left( 592 self, predicate: F1[[Y], bool] 593 ) -> "Xresult[Y, XresultError] | Xresult[Never, X]": 594 """Return a new Xresult with the branch = RIGHT if the predicate is not met. 595 596 Fill the result with a default error mentioning the initial value in case of branch switching. 597 598 ### Return 599 600 - if branch != LEFT -- self 601 - if branch == LEFT and predicate -- self 602 - otherwise -- a new Xresult, having branch = RIGHT, with value = XresultError(self) 603 604 ### Usage 605 606 see filter_right 607 """ 608 match self.map_left(predicate): 609 case Xresult(True, XRBranch.LEFT) | Xresult(_, XRBranch.RIGHT): 610 return cast(Xresult[Never, X], self) 611 case _: 612 return Xresult[Y, XresultError](XresultError(self), XRBranch.RIGHT) 613 614 def filter_right( 615 self, predicate: F1[[X], bool] 616 ) -> "Xresult[Y, Never] | Xresult[XresultError, X]": 617 """Return a new Xresult with the branch = LEFT if the predicate is not met. 618 619 Fill the result with a default error mentioning the initial value in case of branch switching. 620 621 ### Return 622 623 - if branch != RIGHT -- self 624 - if branch == RIGHT and predicate -- self 625 - otherwise -- a new Xresult, having branch = LEFT, with value = XresultError(self) 626 627 ### Usage 628 629 ```python 630 631 ( 632 Xresult 633 .right(4) # Xresult(4, XRBranch.RIGHT) 634 .filter(lambda x: x < 10) # Xresult(4, XRBranch.RIGHT) 635 .filter(lambda x: x > 10) # Xresult( 636 # XresultError(Xresult(4,XRBranch.RIGHT)), 637 # XRBranch.LEFT 638 # ) 639 .filter(lambda _: True) # no change, we are now on the LEFT branch 640 ) 641 ``` 642 """ 643 match self.map_right(predicate): 644 case Xresult(True, XRBranch.RIGHT) | Xresult(_, XRBranch.LEFT): 645 return cast(Xresult[Y, Never], self) 646 case _: 647 return Xresult[XresultError, X](XresultError(self), XRBranch.LEFT)
Encapsulate Union type in container.
Semantically, Xresult helps managing unpure types, such as :
- nullable values (Xresult[None, X])
- tryable values (Xresult[Error, X])
Eventually using it for plain encapsulated Union type is fine.
Attributes
- branch -- XRBranch (LEFT, RIGHT), semaphore telling if the value attribute is an X (LEFT) or a Y (RIGHT)
- value -- Y | X, the content of the container
Features
- Monadic behavior
- Manual handling of the bias
- Class method to lift values to Xresult
Helper classes
Helper classes are provided to semantically instantiate / pattern match your Xresults. See Xeither, Xtry, Xopt for more information.
Usages
- directly returns Xresult in your own functions:
def f_that_breaks(should_break: bool) -> Xresult[Exception, str]:
if should_break:
return Xresult(Exception("something went wrong"), XRBranch.LEFT)
else:
return Xresult("Everything's fine", XRBranch.RIGHT)
- catch common functions into Xresult:
effect_result: Xresult[Exception, E] = Xtry.from_unsafe(some_function_that_raises)
- powerful optional handling:
optional_value: int | None = 3
option_value: Xresult[None, int] = Xopt.from_optional(optional_value)
- rich union type:
def returns_a_str_or_int(should_be_str: boolean) -> Xresult[str, int]:
if should_be_str:
return Xresult("foo", XRBranch.LEFT)
else:
return Xresult(42, XRBranch.RIGHT)
Xresult(value: Union[Y, X], branch: XRBranch)
branch: XRBranch
@staticmethod
def
fors(els: 'F0[list[T]]') -> 'Xresult[XresultError, T]':
127 @staticmethod 128 def fors[T](els: "F0[list[T]]") -> "Xresult[XresultError, T]": 129 """Return the Xresult computed in a list comprehension of zipped Xresult. 130 131 Used as a complement of __iter__ to compose multiple results together. 132 133 ### Keyword Arguments 134 135 - els: lazy list of zero or one element. To be mecanically useful, should be computed as a list comprehension. 136 137 ### Usage 138 139 ```python 140 # Return Xresult(6, XRBranch.RIGHT) 141 Xresult.fors(lambda: # lambda to make the computation lazy 142 [ 143 x + y + z # put here your algorithm for composing effect results 144 for x, y, z 145 in zip( # Chain your effects results in a zip 146 Xeither.Right(1), 147 Xeither.Right(2), 148 Xeither.Right(3) 149 ) 150 ] 151 ) 152 153 # If at least one XResult has a different branch than requested, stop at the first encountered 154 # Return Xresult(2, XRBranch.LEFT) 155 Xresult.fors(lambda: 156 [ 157 x + y + z 158 for x, y, z 159 in zip( 160 Xeither.Right(1), 161 Xeither.Left(2), 162 Xeither.Right(3) 163 ) 164 ] 165 ) 166 ``` 167 """ 168 try: 169 if isinstance(value := els()[0], Xresult): 170 return value 171 else: 172 return Xresult(value, XRBranch.RIGHT) 173 except XresultError as e: 174 return Xresult(e, XRBranch.LEFT)
Return the Xresult computed in a list comprehension of zipped Xresult.
Used as a complement of __iter__ to compose multiple results together.
Keyword Arguments
- els: lazy list of zero or one element. To be mecanically useful, should be computed as a list comprehension.
Usage
# Return Xresult(6, XRBranch.RIGHT)
Xresult.fors(lambda: # lambda to make the computation lazy
[
x + y + z # put here your algorithm for composing effect results
for x, y, z
in zip( # Chain your effects results in a zip
Xeither.Right(1),
Xeither.Right(2),
Xeither.Right(3)
)
]
)
# If at least one XResult has a different branch than requested, stop at the first encountered
# Return Xresult(2, XRBranch.LEFT)
Xresult.fors(lambda:
[
x + y + z
for x, y, z
in zip(
Xeither.Right(1),
Xeither.Left(2),
Xeither.Right(3)
)
]
)
def
map(self, f: F1[[X], T]) -> 'Xresult[Y, T]':
176 def map[T](self, f: F1[[X], T]) -> "Xresult[Y, T]": 177 """Alias for map_right.""" 178 return self.map_right(f)
Alias for map_right.
def
map_left(self, f: F1[[Y], U]) -> 'Xresult[U, X]':
180 def map_left[U](self, f: F1[[Y], U]) -> "Xresult[U, X]": 181 """Return either itself or a new Xresult (LEFT) containing the result of f. 182 183 Is mainly used to chain effect free operations. 184 185 ### Return 186 187 - if self is a RIGHT -- self 188 - if self is a LEFT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value) 189 190 ### Usage 191 192 see map_right 193 """ 194 match self: 195 case Xresult(value, XRBranch.LEFT): 196 return Xresult[U, Never](f(cast(Y, value)), XRBranch.LEFT) 197 case _: 198 return cast(Xresult[Never, X], self)
Return either itself or a new Xresult (LEFT) containing the result of f.
Is mainly used to chain effect free operations.
Return
- if self is a RIGHT -- self
- if self is a LEFT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value)
Usage
see map_right
def
map_right(self, f: F1[[X], T]) -> 'Xresult[Y, T]':
200 def map_right[T](self, f: F1[[X], T]) -> "Xresult[Y, T]": 201 """Return either itself or a new Xresult (RIGHT) containing the result of f. 202 203 ### Return 204 205 - if self is a LEFT -- self 206 - if self is a RIGHT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value) 207 208 ### Usage 209 210 ```python 211 def add_three(i: float) -> float: 212 return i + 3 213 214 def pow(i: float) -> float: 215 return i * i 216 217 ( 218 Xopt 219 .from_optional(3) # Xresult(3, XRBranch.RIGHT) 220 .map_right(add_three) # Xresult(6, XRBranch.RIGHT) 221 .map_right(pow) # Xresult(36, XRBranch.RIGHT) 222 .map_right(lambda x: x - 4) # Xresult(32, XRBranch.RIGHT) 223 ) 224 ``` 225 """ 226 match self: 227 case Xresult(value, XRBranch.RIGHT): 228 return Xresult[Never, T](f(cast(X, value)), XRBranch.RIGHT) 229 case _: 230 return cast(Xresult[Y, Never], self)
Return either itself or a new Xresult (RIGHT) containing the result of f.
Return
- if self is a LEFT -- self
- if self is a RIGHT -- a new Xresult, being a copy of the current result with the underlying value = f(self.value)
Usage
def add_three(i: float) -> float:
return i + 3
def pow(i: float) -> float:
return i * i
(
Xopt
.from_optional(3) # Xresult(3, XRBranch.RIGHT)
.map_right(add_three) # Xresult(6, XRBranch.RIGHT)
.map_right(pow) # Xresult(36, XRBranch.RIGHT)
.map_right(lambda x: x - 4) # Xresult(32, XRBranch.RIGHT)
)
def
flatten( self: 'Xresult[YS, Xresult[U, XS]]') -> 'Xresult[YS, Never] | Xresult[U, XS]':
232 def flatten[YS, U, XS]( 233 self: "Xresult[YS, Xresult[U, XS]]", 234 ) -> "Xresult[YS, Never] | Xresult[U, XS]": 235 """Alias for flatten_right.""" 236 return self.flatten_right()
Alias for flatten_right.
def
flatten_left( self: 'Xresult[Xresult[YS, T], XS]') -> 'Xresult[YS, T] | Xresult[Never, XS]':
238 def flatten_left[YS, T, XS]( 239 self: "Xresult[Xresult[YS, T], XS]", 240 ) -> "Xresult[YS, T] | Xresult[Never, XS]": 241 """Return either self or a new flat Xresult if the underlying value is an Xresult. 242 243 ### Return 244 245 - if self.value is an Xresult, and branch = LEFT -- a new Xresult being the underlying value 246 - otherwise -- self 247 248 ### Usage 249 250 see flatten_right 251 """ 252 match self: 253 case Xresult(value, XRBranch.LEFT): 254 return cast(Xresult[YS, T], value) 255 case _: 256 return cast(Xresult[Never, XS], self)
Return either self or a new flat Xresult if the underlying value is an Xresult.
Return
- if self.value is an Xresult, and branch = LEFT -- a new Xresult being the underlying value
- otherwise -- self
Usage
see flatten_right
def
flatten_right( self: 'Xresult[YS, Xresult[U, XS]]') -> 'Xresult[YS, Never] | Xresult[U, XS]':
258 def flatten_right[YS, U, XS]( 259 self: "Xresult[YS, Xresult[U, XS]]", 260 ) -> "Xresult[YS, Never] | Xresult[U, XS]": 261 """Return either self or a new flat Xresult if the underlying value is an Xresult. 262 263 ### Return 264 265 - if self.value is an Xresult, and branch = RIGHT -- a new Xresult being the underlying value 266 - otherwise -- self 267 268 ### Usage 269 270 ```python 271 assert Xresult.right(Xresult.right("example")).flatten_right() == Xresult("example", XRBranch.RIGHT) 272 assert Xresult.right("example").flatten_right() == Xresult("example", XRBranch.RIGHT) 273 assert Xresult.right(Xopt.from_optional(None)).flatten_right() == Xresult(None, XRBranch.LEFT) 274 assert Xresult.right(Xresult.left("example")).flatten_right() == Xresult("example", XRBranch.LEFT) 275 ``` 276 """ 277 match self: 278 case Xresult(value, XRBranch.RIGHT): 279 return cast(Xresult[U, XS], value) 280 case _: 281 return cast(Xresult[YS, Never], self)
Return either self or a new flat Xresult if the underlying value is an Xresult.
Return
- if self.value is an Xresult, and branch = RIGHT -- a new Xresult being the underlying value
- otherwise -- self
Usage
assert Xresult.right(Xresult.right("example")).flatten_right() == Xresult("example", XRBranch.RIGHT)
assert Xresult.right("example").flatten_right() == Xresult("example", XRBranch.RIGHT)
assert Xresult.right(Xopt.from_optional(None)).flatten_right() == Xresult(None, XRBranch.LEFT)
assert Xresult.right(Xresult.left("example")).flatten_right() == Xresult("example", XRBranch.LEFT)
def
flat_map(self, f: 'F1[[X], Xresult[U, T]]') -> 'Xresult[Y, Never] | Xresult[U, T]':
283 def flat_map[T, U]( 284 self, f: "F1[[X], Xresult[U, T]]" 285 ) -> "Xresult[Y, Never] | Xresult[U, T]": 286 """Alias for flat_map_right.""" 287 return self.flat_map_right(f)
Alias for flat_map_right.
def
flat_map_left(self, f: 'F1[[Y], Xresult[U, T]]') -> 'Xresult[U, T] | Xresult[Never, X]':
289 def flat_map_left[T, U]( 290 self, f: "F1[[Y], Xresult[U, T]]" 291 ) -> "Xresult[U, T] | Xresult[Never, X]": 292 """Return the result of map_left then flatten. 293 294 ### Return 295 296 - if self is a RIGHT -- self 297 - if self is a LEFT -- a new Xresult, map_left then flatten 298 299 ### Usage 300 301 see flat_map 302 """ 303 return self.map_left(f).flatten_left()
Return the result of map_left then flatten.
Return
- if self is a RIGHT -- self
- if self is a LEFT -- a new Xresult, map_left then flatten
Usage
see flat_map
def
flat_map_right(self, f: 'F1[[X], Xresult[U, T]]') -> 'Xresult[Y, Never] | Xresult[U, T]':
305 def flat_map_right[T, U]( 306 self, f: "F1[[X], Xresult[U, T]]" 307 ) -> "Xresult[Y, Never] | Xresult[U, T]": 308 """Return the result of map_right then flatten. 309 310 ### Return 311 312 - if self is a LEFT -- self 313 - if self is a RIGHT -- a new Xresult, map_right then flatten 314 315 ### Usage 316 317 ```python 318 from xfp import Xtry 319 import math 320 321 @Xtry.safed 322 def invert(i: float) -> float: 323 return 1 / i 324 325 @Xtry.safed 326 def sqrt(i: float) -> float: 327 return math.sqrt(i) 328 329 ( 330 Xtry 331 .Success(4) # Xresult(4, XRBranch.RIGHT) 332 .flat_map_right(invert) # Xresult(0.25, XRBranch.RIGHT) 333 .flat_map_right(sqrt) # Xresult(0.5, XRBranch.RIGHT) 334 ) 335 ( 336 Xtry 337 .Success(0) # Xresult(0, XRBranch.RIGHT) 338 .flat_map_right(invert) # Xresult(ZeroDivisionError(..., XRBranch.LEFT)) 339 .flat_map_right(sqrt) # Xresult(ZeroDivisionError(..., XRBranch.LEFT)) 340 ) 341 ( 342 Xtry 343 .Success(-4) # Xresult(-4, XRBranch.RIGHT) 344 .flat_map_right(invert) # Xresult(-0.25, XRBranch.RIGHT) 345 .flat_map_right(sqrt) # Xresult(ValueError(..., XRBranch.LEFT)) 346 ) 347 """ 348 return self.map_right(f).flatten_right()
Return the result of map_right then flatten.
Return
- if self is a LEFT -- self
- if self is a RIGHT -- a new Xresult, map_right then flatten
Usage
```python from xfp import Xtry import math
@Xtry.safed
def invert(i: float) -> float:
return 1 / i
@Xtry.safed
def sqrt(i: float) -> float:
return math.sqrt(i)
(
Xtry
.Success(4) # Xresult(4, XRBranch.RIGHT)
.flat_map_right(invert) # Xresult(0.25, XRBranch.RIGHT)
.flat_map_right(sqrt) # Xresult(0.5, XRBranch.RIGHT)
)
(
Xtry
.Success(0) # Xresult(0, XRBranch.RIGHT)
.flat_map_right(invert) # Xresult(ZeroDivisionError(..., XRBranch.LEFT))
.flat_map_right(sqrt) # Xresult(ZeroDivisionError(..., XRBranch.LEFT))
)
(
Xtry
.Success(-4) # Xresult(-4, XRBranch.RIGHT)
.flat_map_right(invert) # Xresult(-0.25, XRBranch.RIGHT)
.flat_map_right(sqrt) # Xresult(ValueError(..., XRBranch.LEFT))
)
def
fold(self: ~Self, x: ~A) -> F1:
Return default if branch != RIGHT, otherwise f(self.value).
Exists as homogenisation with Xlist.fold
Keyword Arguments
- default -- output when the value does not exist on the RIGHT side
- f -- transformation to apply to the underlying value before returning when the value is present on the RIGHT side
Usage
from xfp import Xtry
from datetime import date
def load_dated_partition(partition_value: date) -> str:
return "actual partition"
@Xtry.safed
def to_date(str_date: str) -> date:
print(date.fromisoformat(str_date))
return date.fromisoformat(str_date)
data = to_date("2024-05-01").fold("default partition")(load_dated_partition)
# good date -> date parsed by to_date() -> fold() calls load_dated_partition
data = to_date("2023-02-29").fold("default partition")(load_dated_partition)
# wrong date -> LEFT bias returned by to_date() -> fold returns "default partition"
def
get_or_else(self, default: T) -> Union[X, T]:
389 def get_or_else[T](self, default: T) -> X | T: 390 """Shorthand for self.fold(default)(id) 391 392 ### Usage 393 ```python 394 from xfp import Xtry 395 from datetime import date 396 397 @Xtry.safed 398 def to_date(str_date: str) -> date: 399 return date.fromisoformat(str_date) 400 401 gotten_date = to_date("2024-05-01").get_or_else(date.today()) 402 else_date = to_date("2023-02-29").get_or_else(date.today()) 403 ``` 404 """ 405 return self.fold(default)(lambda x: x)
Shorthand for self.fold(default)(id)
Usage
from xfp import Xtry
from datetime import date
@Xtry.safed
def to_date(str_date: str) -> date:
return date.fromisoformat(str_date)
gotten_date = to_date("2024-05-01").get_or_else(date.today())
else_date = to_date("2023-02-29").get_or_else(date.today())
def
foreach(self, statement: F1[[X], typing.Any]) -> None:
407 def foreach(self, statement: F1[[X], Any]) -> None: 408 """Alias for foreach_right.""" 409 self.foreach_right(statement)
Alias for foreach_right.
def
foreach_left(self, statement: F1[[Y], typing.Any]) -> None:
411 def foreach_left(self, statement: F1[[Y], Any]) -> None: 412 """Do the statement procedure to the underlying value if self is a LEFT. 413 414 ### Usage 415 416 see foreach_right 417 """ 418 if self.branch == XRBranch.LEFT: 419 statement(cast(Y, self.value))
Do the statement procedure to the underlying value if self is a LEFT.
Usage
see foreach_right
def
foreach_right(self, statement: F1[[X], typing.Any]) -> None:
421 def foreach_right(self, statement: F1[[X], Any]) -> None: 422 """Do the statement procedure to the underlying value if self is a RIGHT. 423 424 ### Usage 425 ```python 426 from xfp import Xresult,XRBranch,Xopt 427 428 ( 429 Xopt 430 .from_optional(25) 431 .foreach(lambda x: print(f"This is an element of the list : {x}")) 432 ) 433 # This is an element of the list : 25 434 435 ( 436 Xopt 437 .from_optional(None) 438 .foreach(lambda x: print(f"This is the element : {x}")) 439 ) 440 # doesn't output anything 441 442 ( 443 Xresult(42, XRBranch.RIGHT) 444 .foreach(lambda x: print(f"This is the right element : {x}")) 445 ) 446 # This is the right element : 42 447 448 ( 449 Xresult(666, XRBranch.LEFT) 450 .foreach(lambda x: print(f"This is the right element : {x}")) 451 ) 452 # doesn't output anything 453 ``` 454 """ 455 if self.branch == XRBranch.RIGHT: 456 statement(cast(X, self.value))
Do the statement procedure to the underlying value if self is a RIGHT.
Usage
from xfp import Xresult,XRBranch,Xopt
(
Xopt
.from_optional(25)
.foreach(lambda x: print(f"This is an element of the list : {x}"))
)
# This is an element of the list : 25
(
Xopt
.from_optional(None)
.foreach(lambda x: print(f"This is the element : {x}"))
)
# doesn't output anything
(
Xresult(42, XRBranch.RIGHT)
.foreach(lambda x: print(f"This is the right element : {x}"))
)
# This is the right element : 42
(
Xresult(666, XRBranch.LEFT)
.foreach(lambda x: print(f"This is the right element : {x}"))
)
# doesn't output anything
def
recover_with(self, f: 'F1[[Y], Xresult[U, T]]') -> 'Xresult[U, T] | Xresult[Never, X]':
458 def recover_with[T, U]( 459 self, f: "F1[[Y], Xresult[U, T]]" 460 ) -> "Xresult[U, T] | Xresult[Never, X]": 461 """Alias for recover_with_right.""" 462 return self.recover_with_right(f)
Alias for recover_with_right.
def
recover_with_left(self, f: 'F1[[X], Xresult[U, T]]') -> 'Xresult[Y, Never] | Xresult[U, T]':
464 def recover_with_left[T, U]( 465 self, f: "F1[[X], Xresult[U, T]]" 466 ) -> "Xresult[Y, Never] | Xresult[U, T]": 467 """Return itself, mapped on the right side, flattened on the left side. 468 469 See flat_map_right 470 471 ### Usage 472 473 See recover_with_right 474 """ 475 match self: 476 case Xresult(v, XRBranch.RIGHT): 477 return f(cast(X, v)) 478 case _: 479 return cast(Xresult[Y, Never], self)
Return itself, mapped on the right side, flattened on the left side.
See flat_map_right
Usage
See recover_with_right
def
recover_with_right(self, f: 'F1[[Y], Xresult[U, T]]') -> 'Xresult[U, T] | Xresult[Never, X]':
481 def recover_with_right[T, U]( 482 self, f: "F1[[Y], Xresult[U, T]]" 483 ) -> "Xresult[U, T] | Xresult[Never, X]": 484 """Return itself, mapped on the left side, flattened on the right side. 485 486 See flat_map_left 487 488 ### Usage 489 490 ```python 491 from xfp import Xtry 492 import math 493 494 @Xtry.safed 495 def invert(i: float) -> float: 496 return 1 / i 497 498 @Xtry.safed 499 def sqrt(i: float) -> float: 500 return math.sqrt(i) 501 502 ( 503 sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT) 504 .recover_with_right(lambda _: invert(-4)) # Xresult(-0.25, XRBranch.RIGHT) 505 ) 506 507 ( 508 sqrt(4) # Xresult(2, XRBranch.RIGHT) 509 .recover_with_right(lambda _: invert(-4)) # Xresult(2, XRBranch.RIGHT) 510 ) 511 ``` 512 """ 513 match self: 514 case Xresult(v, XRBranch.LEFT): 515 return f(cast(Y, v)) 516 case _: 517 return cast(Xresult[Never, X], self)
Return itself, mapped on the left side, flattened on the right side.
See flat_map_left
Usage
from xfp import Xtry
import math
@Xtry.safed
def invert(i: float) -> float:
return 1 / i
@Xtry.safed
def sqrt(i: float) -> float:
return math.sqrt(i)
(
sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT)
.recover_with_right(lambda _: invert(-4)) # Xresult(-0.25, XRBranch.RIGHT)
)
(
sqrt(4) # Xresult(2, XRBranch.RIGHT)
.recover_with_right(lambda _: invert(-4)) # Xresult(2, XRBranch.RIGHT)
)
def
recover(self, f: F1[[Y], T]) -> 'Xresult[Never, X | T]':
519 def recover[T](self, f: F1[[Y], T]) -> "Xresult[Never, X | T]": 520 """Alias for recover_right.""" 521 return self.recover_right(f)
Alias for recover_right.
def
recover_left(self, f: F1[[X], U]) -> 'Xresult[Y | U, Never]':
523 def recover_left[U](self, f: F1[[X], U]) -> "Xresult[Y | U, Never]": 524 """Return a new Xresult with is always a LEFT. 525 526 Used to convert a RIGHT result into a LEFT using an effectless transformation. 527 Semantically : 528 Used to fallback from a potential success with an effectless operation. 529 This is a fallback that always ends up as a 'failure'. 530 531 ### Return 532 533 - if branch == LEFT -- self 534 - otherwise -- a new Xresult, having branch = LEFT, with an inherent value of f(self.value) 535 536 ### Usage 537 538 See recover_right 539 """ 540 match self: 541 case Xresult(v, XRBranch.RIGHT): 542 return Xresult(f(cast(X, v)), XRBranch.LEFT) 543 case _: 544 return cast(Xresult[Y, Never], self)
Return a new Xresult with is always a LEFT.
Used to convert a RIGHT result into a LEFT using an effectless transformation. Semantically : Used to fallback from a potential success with an effectless operation. This is a fallback that always ends up as a 'failure'.
Return
- if branch == LEFT -- self
- otherwise -- a new Xresult, having branch = LEFT, with an inherent value of f(self.value)
Usage
See recover_right
def
recover_right(self, f: F1[[Y], T]) -> 'Xresult[Never, T | X]':
546 def recover_right[T](self, f: F1[[Y], T]) -> "Xresult[Never, T | X]": 547 """Return a new Xresult with is always a RIGHT. 548 549 Used to convert a LEFT result into a RIGHT using an effectless transformation. 550 Semantically : 551 Used to fallback from a potential failure with an effectless operation. 552 This is a fallback that always ends up as a 'success'. 553 554 ### Return 555 556 - if branch == RIGHT -- self 557 - otherwise -- a new Xresult, having branch = RIGHT, with an inherent value of f(self.value) 558 559 ### Usage 560 561 ```python 562 from xfp import Xtry 563 import math 564 565 @Xtry.safed 566 def sqrt(i: float) -> float: 567 return math.sqrt(i) 568 569 ( 570 sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT) 571 .recover_right(lambda _: 0) # Xresult(0, XRBranch.RIGHT) 572 ) 573 ( 574 sqrt(4) # Xresult(2, XRBranch.RIGHT) 575 .recover_right(lambda _: 0) # Xresult(2.0, XRBranch.RIGHT) 576 ) 577 ``` 578 """ 579 match self: 580 case Xresult(v, XRBranch.LEFT): 581 return Xresult(f(cast(Y, v)), XRBranch.RIGHT) 582 case _: 583 return cast(Xresult[Never, X], self)
Return a new Xresult with is always a RIGHT.
Used to convert a LEFT result into a RIGHT using an effectless transformation. Semantically : Used to fallback from a potential failure with an effectless operation. This is a fallback that always ends up as a 'success'.
Return
- if branch == RIGHT -- self
- otherwise -- a new Xresult, having branch = RIGHT, with an inherent value of f(self.value)
Usage
from xfp import Xtry
import math
@Xtry.safed
def sqrt(i: float) -> float:
return math.sqrt(i)
(
sqrt(-4) # Xresult(ValueError(...), XRBranch.LEFT)
.recover_right(lambda _: 0) # Xresult(0, XRBranch.RIGHT)
)
(
sqrt(4) # Xresult(2, XRBranch.RIGHT)
.recover_right(lambda _: 0) # Xresult(2.0, XRBranch.RIGHT)
)
def
filter( self, predicate: F1[[X], bool]) -> 'Xresult[Y, Never] | Xresult[XresultError, X]':
585 def filter( 586 self, predicate: F1[[X], bool] 587 ) -> "Xresult[Y, Never] | Xresult[XresultError, X]": 588 """Alias for filter_right.""" 589 return self.filter_right(predicate)
Alias for filter_right.
def
filter_left( self, predicate: F1[[Y], bool]) -> 'Xresult[Y, XresultError] | Xresult[Never, X]':
591 def filter_left( 592 self, predicate: F1[[Y], bool] 593 ) -> "Xresult[Y, XresultError] | Xresult[Never, X]": 594 """Return a new Xresult with the branch = RIGHT if the predicate is not met. 595 596 Fill the result with a default error mentioning the initial value in case of branch switching. 597 598 ### Return 599 600 - if branch != LEFT -- self 601 - if branch == LEFT and predicate -- self 602 - otherwise -- a new Xresult, having branch = RIGHT, with value = XresultError(self) 603 604 ### Usage 605 606 see filter_right 607 """ 608 match self.map_left(predicate): 609 case Xresult(True, XRBranch.LEFT) | Xresult(_, XRBranch.RIGHT): 610 return cast(Xresult[Never, X], self) 611 case _: 612 return Xresult[Y, XresultError](XresultError(self), XRBranch.RIGHT)
Return a new Xresult with the branch = RIGHT if the predicate is not met.
Fill the result with a default error mentioning the initial value in case of branch switching.
Return
- if branch != LEFT -- self
- if branch == LEFT and predicate -- self
- otherwise -- a new Xresult, having branch = RIGHT, with value = XresultError(self)
Usage
see filter_right
def
filter_right( self, predicate: F1[[X], bool]) -> 'Xresult[Y, Never] | Xresult[XresultError, X]':
614 def filter_right( 615 self, predicate: F1[[X], bool] 616 ) -> "Xresult[Y, Never] | Xresult[XresultError, X]": 617 """Return a new Xresult with the branch = LEFT if the predicate is not met. 618 619 Fill the result with a default error mentioning the initial value in case of branch switching. 620 621 ### Return 622 623 - if branch != RIGHT -- self 624 - if branch == RIGHT and predicate -- self 625 - otherwise -- a new Xresult, having branch = LEFT, with value = XresultError(self) 626 627 ### Usage 628 629 ```python 630 631 ( 632 Xresult 633 .right(4) # Xresult(4, XRBranch.RIGHT) 634 .filter(lambda x: x < 10) # Xresult(4, XRBranch.RIGHT) 635 .filter(lambda x: x > 10) # Xresult( 636 # XresultError(Xresult(4,XRBranch.RIGHT)), 637 # XRBranch.LEFT 638 # ) 639 .filter(lambda _: True) # no change, we are now on the LEFT branch 640 ) 641 ``` 642 """ 643 match self.map_right(predicate): 644 case Xresult(True, XRBranch.RIGHT) | Xresult(_, XRBranch.LEFT): 645 return cast(Xresult[Y, Never], self) 646 case _: 647 return Xresult[XresultError, X](XresultError(self), XRBranch.LEFT)
Return a new Xresult with the branch = LEFT if the predicate is not met.
Fill the result with a default error mentioning the initial value in case of branch switching.
Return
- if branch != RIGHT -- self
- if branch == RIGHT and predicate -- self
- otherwise -- a new Xresult, having branch = LEFT, with value = XresultError(self)
Usage
(
Xresult
.right(4) # Xresult(4, XRBranch.RIGHT)
.filter(lambda x: x < 10) # Xresult(4, XRBranch.RIGHT)
.filter(lambda x: x > 10) # Xresult(
# XresultError(Xresult(4,XRBranch.RIGHT)),
# XRBranch.LEFT
# )
.filter(lambda _: True) # no change, we are now on the LEFT branch
)
class
XRBranch(enum.Enum):
8class XRBranch(Enum): 9 LEFT = auto() 10 RIGHT = auto() 11 12 def invert(self) -> "XRBranch": 13 return XRBranch.RIGHT if self == XRBranch.LEFT else XRBranch.LEFT
LEFT =
<XRBranch.LEFT: 1>
RIGHT =
<XRBranch.RIGHT: 2>
Inherited Members
- enum.Enum
- name
- value
@dataclass(init=False)
class
XresultError16@dataclass(init=False) 17class XresultError[Y, X](Exception): 18 xresult: "Xresult[Y, X]" 19 20 def __init__(self, xresult: "Xresult[Y, X]"): 21 self.xresult = xresult 22 super().__init__(f"Auto generated error for initial result {self.xresult}")
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- add_note
- args
class
Xeither:
7class Xeither: 8 """Common tools to instantiate and pattern match Xresult. 9 10 ### Provides 11 12 Proxy classes to quickly instantiate LEFT and RIGHT. 13 Approximately makes Xresult(..., XRBranch.LEFT) and Xresult(..., XRBranch.RIGHT) independent classes. 14 15 ### Usage 16 17 ```python 18 regular_result: Xresult[None, Any] = Xresult(None, XRBranch.LEFT) 19 either_result: Xresult[Any, Int] = Xeither.Right(3) 20 21 match either_result: 22 case Xeither.Right(value): 23 print(value) 24 case Xeither.Left(_): 25 print("no value") 26 27 # You can also pattern match regular Xresult with Left/Right 28 match regular_result: 29 case Xeither.Right(value): 30 print(value) 31 case Xeither.Left(_): 32 print("no value") 33 ``` 34 """ 35 36 class XresultLeftMeta(type): 37 """Metaclass to interprets Left as an Xresult[Y, Never]. 38 39 Overrides the instanceof in order to enable pattern matching between Left and Xresult. 40 """ 41 42 def __instancecheck__(self, instance): 43 return isinstance(instance, Xresult) and instance.branch == XRBranch.LEFT 44 45 @dataclass(frozen=True, init=False, match_args=False, eq=False) 46 class Left[Y](Xresult[Y, Never], metaclass=XresultLeftMeta): 47 """Specific result holding a LEFT. 48 49 Can also act as an extractor in pattern matching. 50 """ 51 52 __match_args__ = ("value",) # type: ignore 53 value: Y 54 55 def __init__(self, value): 56 super().__init__(value, XRBranch.LEFT) 57 58 class XresultRightMeta(type): 59 """Metaclass to interprets Right as an Xresult[Never, X]. 60 61 Overrides the instanceof in order to enable pattern matching between Right and Xresult. 62 """ 63 64 def __instancecheck__(self, instance): 65 return isinstance(instance, Xresult) and instance.branch == XRBranch.RIGHT 66 67 @dataclass(frozen=True, init=False, match_args=False, eq=False) 68 class Right[X](Xresult[Never, X], metaclass=XresultRightMeta): 69 """Specific result holding a RIGHT. 70 71 Can also act as an extractor in pattern matching. 72 """ 73 74 __match_args__ = ("value",) # type: ignore 75 value: X 76 77 def __init__(self, value): 78 super().__init__(value, XRBranch.RIGHT)
Common tools to instantiate and pattern match Xresult.
Provides
Proxy classes to quickly instantiate LEFT and RIGHT. Approximately makes Xresult(..., XRBranch.LEFT) and Xresult(..., XRBranch.RIGHT) independent classes.
Usage
regular_result: Xresult[None, Any] = Xresult(None, XRBranch.LEFT)
either_result: Xresult[Any, Int] = Xeither.Right(3)
match either_result:
case Xeither.Right(value):
print(value)
case Xeither.Left(_):
print("no value")
# You can also pattern match regular Xresult with Left/Right
match regular_result:
case Xeither.Right(value):
print(value)
case Xeither.Left(_):
print("no value")
class
Xeither.XresultLeftMeta(builtins.type):
36 class XresultLeftMeta(type): 37 """Metaclass to interprets Left as an Xresult[Y, Never]. 38 39 Overrides the instanceof in order to enable pattern matching between Left and Xresult. 40 """ 41 42 def __instancecheck__(self, instance): 43 return isinstance(instance, Xresult) and instance.branch == XRBranch.LEFT
Metaclass to interprets Left as an Xresult[Y, Never].
Overrides the instanceof in order to enable pattern matching between Left and Xresult.
Inherited Members
- builtins.type
- type
- mro
@dataclass(frozen=True, init=False, match_args=False, eq=False)
class
Xeither.Left45 @dataclass(frozen=True, init=False, match_args=False, eq=False) 46 class Left[Y](Xresult[Y, Never], metaclass=XresultLeftMeta): 47 """Specific result holding a LEFT. 48 49 Can also act as an extractor in pattern matching. 50 """ 51 52 __match_args__ = ("value",) # type: ignore 53 value: Y 54 55 def __init__(self, value): 56 super().__init__(value, XRBranch.LEFT)
Specific result holding a LEFT.
Can also act as an extractor in pattern matching.
Inherited Members
class
Xeither.XresultRightMeta(builtins.type):
58 class XresultRightMeta(type): 59 """Metaclass to interprets Right as an Xresult[Never, X]. 60 61 Overrides the instanceof in order to enable pattern matching between Right and Xresult. 62 """ 63 64 def __instancecheck__(self, instance): 65 return isinstance(instance, Xresult) and instance.branch == XRBranch.RIGHT
Metaclass to interprets Right as an Xresult[Never, X].
Overrides the instanceof in order to enable pattern matching between Right and Xresult.
Inherited Members
- builtins.type
- type
- mro
@dataclass(frozen=True, init=False, match_args=False, eq=False)
class
Xeither.Right67 @dataclass(frozen=True, init=False, match_args=False, eq=False) 68 class Right[X](Xresult[Never, X], metaclass=XresultRightMeta): 69 """Specific result holding a RIGHT. 70 71 Can also act as an extractor in pattern matching. 72 """ 73 74 __match_args__ = ("value",) # type: ignore 75 value: X 76 77 def __init__(self, value): 78 super().__init__(value, XRBranch.RIGHT)
Specific result holding a RIGHT.
Can also act as an extractor in pattern matching.
Inherited Members
class
Xopt:
8class Xopt: 9 """Common tools to instantiate and pattern match Xresult[None, X]. 10 11 ### Provides 12 13 - methods to formalize Optional values 14 - value Empty and type Some, interpretables as Xresult, and usable as Xresult pattern matching 15 16 ### Usage 17 18 ```python 19 regular_result: Xresult[None, Any] = Xresult(None, XRBranch.LEFT) 20 optional_result: Xresult[None, Int] = Xopt.from_optional(3) 21 22 match optional_result: 23 case Xopt.Some(value): 24 print(value) 25 case Xopt.Empty: 26 print("no value") 27 28 # You can also pattern match regular Xresult with Some/Empty 29 match regular_result: 30 case Xopt.Some(value): 31 print(value) 32 case Xopt.Empty: 33 print("no value") 34 ``` 35 """ 36 37 @classmethod 38 def from_optional[X](cls, value: None | X) -> Xresult[None, X]: 39 """Return an Xresult from an optional value. 40 41 value: 42 - X -- Return a Xopt.Some 43 - None -- Return Xopt.Empty 44 """ 45 match value: 46 case None: 47 return cls.Empty 48 case _: 49 return cls.Some(value) 50 51 Empty: Xresult[None, Never] = Xeither.Left(None) 52 53 class XresultMeta(type): 54 """Metaclass to interprets Some as an Xresult[None, X]. 55 56 Overrides the instanceof in order to enable pattern matching between Some and Xresult. 57 """ 58 59 def __instancecheck__(self, instance): 60 return isinstance(instance, Xeither.Right) 61 62 @dataclass(frozen=True, init=False, match_args=False, eq=False) 63 class Some[X](Xresult[Never, X], metaclass=XresultMeta): 64 """Specific result holding a value, with alternate path being None. 65 66 Can also act as an extractor in pattern matching. 67 """ 68 69 __match_args__ = ("value",) # type: ignore 70 value: X 71 72 def __init__(self, value): 73 super().__init__(value, XRBranch.RIGHT)
Common tools to instantiate and pattern match Xresult[None, X].
Provides
- methods to formalize Optional values
- value Empty and type Some, interpretables as Xresult, and usable as Xresult pattern matching
Usage
regular_result: Xresult[None, Any] = Xresult(None, XRBranch.LEFT)
optional_result: Xresult[None, Int] = Xopt.from_optional(3)
match optional_result:
case Xopt.Some(value):
print(value)
case Xopt.Empty:
print("no value")
# You can also pattern match regular Xresult with Some/Empty
match regular_result:
case Xopt.Some(value):
print(value)
case Xopt.Empty:
print("no value")
class
Xopt.XresultMeta(builtins.type):
53 class XresultMeta(type): 54 """Metaclass to interprets Some as an Xresult[None, X]. 55 56 Overrides the instanceof in order to enable pattern matching between Some and Xresult. 57 """ 58 59 def __instancecheck__(self, instance): 60 return isinstance(instance, Xeither.Right)
Metaclass to interprets Some as an Xresult[None, X].
Overrides the instanceof in order to enable pattern matching between Some and Xresult.
Inherited Members
- builtins.type
- type
- mro
@dataclass(frozen=True, init=False, match_args=False, eq=False)
class
Xopt.Some62 @dataclass(frozen=True, init=False, match_args=False, eq=False) 63 class Some[X](Xresult[Never, X], metaclass=XresultMeta): 64 """Specific result holding a value, with alternate path being None. 65 66 Can also act as an extractor in pattern matching. 67 """ 68 69 __match_args__ = ("value",) # type: ignore 70 value: X 71 72 def __init__(self, value): 73 super().__init__(value, XRBranch.RIGHT)
Specific result holding a value, with alternate path being None.
Can also act as an extractor in pattern matching.
Inherited Members
class
Xtry:
12class Xtry: 13 """Common tools to instantiate and pattern match Xresult[Exception, X]. 14 15 ### Provides 16 17 - methods to secure failable functions 18 - types Success and Failure, instantiable as Xresult, and usable as Xresult pattern matching 19 20 ### Usage 21 22 ```python 23 regular_result: Xresult[Exception, Any] = Xresult(Exception("Something went wrong"), XRBranch.LEFT) 24 try_result: Xresult[Exception, Int] = Xtry.Success(3) 25 26 match try_result: 27 case Xtry.Success(value): 28 print(value) 29 case Xtry.Failure(e): 30 raise e 31 32 # You can also pattern match regular Xresult with Success/Failure 33 match regular_result: 34 case Xtry.Success(value): 35 print(value) 36 case Xtry.Failure(e): 37 raise e 38 ``` 39 """ 40 41 @classmethod 42 def from_unsafe[X](cls, f: F0[X]) -> Xresult[Exception, X]: 43 """Return the result of a function as an Xresult. 44 45 Execute the callable and catch the result : 46 - Callable returns -- wrap the result in a RIGHT 47 - Callable raises -- wrap the Exception in a LEFT 48 49 ### Usage 50 51 ```python 52 def unsafe_function(param: str) -> int: 53 if param == "": 54 raise Exception("error") 55 return 3 56 57 a: Xresult[Exception, int] = Xtry.from_unsafe(lambda: unsafe_function("foo")) 58 ``` 59 """ 60 try: 61 return cls.Success(f()) 62 except Exception as e: 63 return cls.Failure(e) 64 65 @classmethod 66 def safed(cls, f: F1[P, X]) -> F1[P, Xresult[Exception, X]]: 67 """Return a new function being f with the side effect wrapped. 68 69 Used as a decorator for quickly converting unsafe code into safe one. 70 Downside is there is no fine tuning over the caught exception. 71 72 ### Usage 73 74 ```python 75 @Xtry.safed 76 def unsafe_function(param: str) -> int: 77 if param == "": 78 raise Exception("error") 79 return 3 80 81 a: Xresult[Exception, int] = unsafe_function("foo") 82 ``` 83 """ 84 85 def inner(*args: P.args, **kwargs: P.kwargs) -> Xresult[Exception, X]: 86 return cls.from_unsafe(lambda: f(*args, **kwargs)) 87 88 return inner 89 90 class XresultFailureMeta(type): 91 """Metaclass to interprets Failure as an Xresult[Exception, Any]. 92 93 Overrides the instanceof in order to enable pattern matching between Failure and Xresult. 94 """ 95 96 def __instancecheck__(self, instance): 97 return isinstance(instance, Xeither.Left) and isinstance( 98 instance.value, Exception 99 ) 100 101 @dataclass(frozen=True, init=False, match_args=False, eq=False) 102 class Failure[Y: Exception](Xresult[Y, Never], metaclass=XresultFailureMeta): 103 """Specific result holding an exception. 104 105 Can also act as an extractor in pattern matching. 106 """ 107 108 __match_args__ = ("value",) # type: ignore 109 value: Y 110 111 def __init__(self, value): 112 super().__init__(value, XRBranch.LEFT) 113 114 class XresultSuccessMeta(type): 115 """Metaclass to interprets Success as an Xresult[Any, X]. 116 117 Overrides the instanceof in order to enable pattern matching between Success and Xresult. 118 """ 119 120 def __instancecheck__(self, instance): 121 return isinstance(instance, Xeither.Right) 122 123 @dataclass(frozen=True, init=False, match_args=False, eq=False) 124 class Success[X](Xresult[Never, X], metaclass=XresultSuccessMeta): 125 """Specific result holding a value, with alternate path being an exception. 126 127 Can also act as an extractor in pattern matching. 128 """ 129 130 __match_args__ = ("value",) # type: ignore 131 value: X 132 133 def __init__(self, value): 134 super().__init__(value, XRBranch.RIGHT)
Common tools to instantiate and pattern match Xresult[Exception, X].
Provides
- methods to secure failable functions
- types Success and Failure, instantiable as Xresult, and usable as Xresult pattern matching
Usage
regular_result: Xresult[Exception, Any] = Xresult(Exception("Something went wrong"), XRBranch.LEFT)
try_result: Xresult[Exception, Int] = Xtry.Success(3)
match try_result:
case Xtry.Success(value):
print(value)
case Xtry.Failure(e):
raise e
# You can also pattern match regular Xresult with Success/Failure
match regular_result:
case Xtry.Success(value):
print(value)
case Xtry.Failure(e):
raise e
41 @classmethod 42 def from_unsafe[X](cls, f: F0[X]) -> Xresult[Exception, X]: 43 """Return the result of a function as an Xresult. 44 45 Execute the callable and catch the result : 46 - Callable returns -- wrap the result in a RIGHT 47 - Callable raises -- wrap the Exception in a LEFT 48 49 ### Usage 50 51 ```python 52 def unsafe_function(param: str) -> int: 53 if param == "": 54 raise Exception("error") 55 return 3 56 57 a: Xresult[Exception, int] = Xtry.from_unsafe(lambda: unsafe_function("foo")) 58 ``` 59 """ 60 try: 61 return cls.Success(f()) 62 except Exception as e: 63 return cls.Failure(e)
Return the result of a function as an Xresult.
Execute the callable and catch the result :
- Callable returns -- wrap the result in a RIGHT
- Callable raises -- wrap the Exception in a LEFT
Usage
def unsafe_function(param: str) -> int:
if param == "":
raise Exception("error")
return 3
a: Xresult[Exception, int] = Xtry.from_unsafe(lambda: unsafe_function("foo"))
65 @classmethod 66 def safed(cls, f: F1[P, X]) -> F1[P, Xresult[Exception, X]]: 67 """Return a new function being f with the side effect wrapped. 68 69 Used as a decorator for quickly converting unsafe code into safe one. 70 Downside is there is no fine tuning over the caught exception. 71 72 ### Usage 73 74 ```python 75 @Xtry.safed 76 def unsafe_function(param: str) -> int: 77 if param == "": 78 raise Exception("error") 79 return 3 80 81 a: Xresult[Exception, int] = unsafe_function("foo") 82 ``` 83 """ 84 85 def inner(*args: P.args, **kwargs: P.kwargs) -> Xresult[Exception, X]: 86 return cls.from_unsafe(lambda: f(*args, **kwargs)) 87 88 return inner
Return a new function being f with the side effect wrapped.
Used as a decorator for quickly converting unsafe code into safe one. Downside is there is no fine tuning over the caught exception.
Usage
@Xtry.safed
def unsafe_function(param: str) -> int:
if param == "":
raise Exception("error")
return 3
a: Xresult[Exception, int] = unsafe_function("foo")
class
Xtry.XresultFailureMeta(builtins.type):
90 class XresultFailureMeta(type): 91 """Metaclass to interprets Failure as an Xresult[Exception, Any]. 92 93 Overrides the instanceof in order to enable pattern matching between Failure and Xresult. 94 """ 95 96 def __instancecheck__(self, instance): 97 return isinstance(instance, Xeither.Left) and isinstance( 98 instance.value, Exception 99 )
Metaclass to interprets Failure as an Xresult[Exception, Any].
Overrides the instanceof in order to enable pattern matching between Failure and Xresult.
Inherited Members
- builtins.type
- type
- mro
@dataclass(frozen=True, init=False, match_args=False, eq=False)
class
Xtry.Failure101 @dataclass(frozen=True, init=False, match_args=False, eq=False) 102 class Failure[Y: Exception](Xresult[Y, Never], metaclass=XresultFailureMeta): 103 """Specific result holding an exception. 104 105 Can also act as an extractor in pattern matching. 106 """ 107 108 __match_args__ = ("value",) # type: ignore 109 value: Y 110 111 def __init__(self, value): 112 super().__init__(value, XRBranch.LEFT)
Specific result holding an exception.
Can also act as an extractor in pattern matching.
Inherited Members
class
Xtry.XresultSuccessMeta(builtins.type):
114 class XresultSuccessMeta(type): 115 """Metaclass to interprets Success as an Xresult[Any, X]. 116 117 Overrides the instanceof in order to enable pattern matching between Success and Xresult. 118 """ 119 120 def __instancecheck__(self, instance): 121 return isinstance(instance, Xeither.Right)
Metaclass to interprets Success as an Xresult[Any, X].
Overrides the instanceof in order to enable pattern matching between Success and Xresult.
Inherited Members
- builtins.type
- type
- mro
@dataclass(frozen=True, init=False, match_args=False, eq=False)
class
Xtry.Success123 @dataclass(frozen=True, init=False, match_args=False, eq=False) 124 class Success[X](Xresult[Never, X], metaclass=XresultSuccessMeta): 125 """Specific result holding a value, with alternate path being an exception. 126 127 Can also act as an extractor in pattern matching. 128 """ 129 130 __match_args__ = ("value",) # type: ignore 131 value: X 132 133 def __init__(self, value): 134 super().__init__(value, XRBranch.RIGHT)
Specific result holding a value, with alternate path being an exception.
Can also act as an extractor in pattern matching.
Inherited Members
class
Xdict(typing.Generic[Y, X]):
34class Xdict[Y, X]: 35 @classmethod 36 def from_list(cls, iterable: ABCIterable[tuple[Y, X]]) -> "Xdict[Y, X]": 37 """Return a new Xdict built from an iterable. 38 39 The iterable must contain couples of <key, values>. In case of key duplication 40 in the parameters list, the last associated value is kept. 41 42 ### Usage 43 44 ```python 45 from xfp import Xdict 46 47 xdict = Xdict.from_list([("a", 1), ("b", 2), ("a", 3)]) 48 assert xdict == Xdict({"a": 3, "b": 2}) 49 ``` 50 """ 51 return cls({k: v for k, v in iterable}) 52 53 def __init__(self, dic: ABCDict[Y, X]) -> None: 54 """Construct an Xdict from a dict-like. 55 56 Dict-like is defined by the existence of the "items" method. 57 """ 58 self.__data = dict(dic.items()) 59 60 def __iter__(self) -> Iterator[tuple[Y, X]]: 61 """Return an iterable over the underlying data.""" 62 return iter(self.items()) 63 64 def __len__(self) -> int: 65 """Return the length of the underlying data. 66 67 Length of an Xdict is the number of distinct keys. 68 """ 69 return len(self.__data) 70 71 def __eq__(self, other: object) -> bool: 72 """Return the equality by comparison of inner values (unordered).""" 73 match other: 74 case ABCDict(): 75 return set([e for e in self.items()]) == set([e for e in other.items()]) 76 case _: 77 return False 78 79 def __repr__(self) -> str: 80 """Return the representation of the underlying data""" 81 return f"Xdict({repr(self.__data)})" 82 83 def __contains__(self, key: Any) -> bool: 84 """Return the presence of the key in the xdict keyset.""" 85 return key in self.keys() 86 87 def __getitem__(self, i: Y) -> X: 88 """Alias for get(i). 89 90 Exists to enable [] syntax 91 92 ### Raises 93 94 - IndexError : if the key is not found in the Xdict 95 """ 96 return self.get(i) 97 98 @overload 99 def get(self, y: Y, /) -> X: 100 """Return the value associated with a given key. 101 102 ### Raises 103 104 - IndexError : if the key is not found in the Xdict 105 """ 106 ... 107 108 @overload 109 def get(self, y: Y, default: X, /) -> X: 110 """Return the value associated with a given key. 111 112 If the key is not found in the Xdict, return the given default instead. 113 """ 114 ... 115 116 def get(self, *args) -> X: 117 """Return the value associated with a given key. 118 119 Implementation for the different `get` methods. 120 Should not be called outside of the provided overloaded signatures. 121 122 ### Raises 123 124 - IndexError : if the key is not found in the Xdict and no default is provided 125 - AttributeError : if the method is called with an unspecified set of parameters (signature not found in overload) 126 """ 127 match args: 128 case [y] if y in self.keys(): 129 return cast(X, self.__data.get(y)) 130 case [y]: 131 raise IndexError(f"Key not found in Xdict : {y}") 132 case [y, default]: 133 return self.__data.get(y, default) 134 case _: 135 raise AttributeError("Wrong set of parameters for <get> method") 136 137 def get_fr(self, y: Y, /) -> Xresult[IndexError, X]: 138 """Return the value associated with a given key. 139 140 Wrap the result in an Xtry to catch potential errors. 141 142 ### Returns 143 144 - Xtry.Success : an Xresult containing the value if the key is found 145 - Xtry.Failure : with an IndexError if the key is not found 146 147 ### Usage 148 149 ```python 150 from xfp import Xdict, Xtry 151 152 match Xdict({"a": 1}).get_fr("b"): 153 case Xtry.Success(value): 154 print(f"we found {value} in Xdict") 155 case Xtry.Failure(e): 156 print(f"Error: {e}") 157 ``` 158 """ 159 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(y))) 160 161 def updated[T](self, key: Y, value: T) -> "Xdict[Y, X | T]": 162 """Return a new Xdict, with an updated couple (key: Y, value: E). 163 164 Upsert a new `value` at `key`. 165 166 ### Usage 167 168 ```python 169 from xfp import Xdict 170 171 assert Xdict({"a": 1}).updated("b", 2) == Xdict({"a": 1, "b": 2}) 172 assert Xdict({"a": 1}).updated("a", 2) == Xdict({"a": 2}) 173 ``` 174 """ 175 return self.union(Xdict({key: value})) 176 177 def removed(self, key: Y) -> "Xdict[Y, X]": 178 """Return a new Xdict, with the given key deleted. 179 180 Filter the provided key if found. 181 No error is raised if the key doesn't exist. 182 183 ### Usage 184 185 ```python 186 from xfp import Xdict 187 188 assert Xdict({"a": 1, "b": 2}).removed("b") == Xdict({"a": 1}) 189 assert Xdict({"a": 1}).removed("b") == Xdict({"a": 1}) 190 ``` 191 """ 192 return self.filter(lambda y, _: y is not key) 193 194 def union[T, U](self, other: ABCDict[U, T]) -> "Xdict[Y | U, X | T]": 195 """Return a new Xdict, being the merge of self and a given one. 196 197 Works as if multiple updateds are done successively. 198 It means if a key is present in both Xdict, the `other` Xdict has priority. 199 200 ### Usage 201 202 ```python 203 from xfp import Xdict 204 205 assert Xdict({"a": 1, "b": 2}).union(Xdict({"a": 3, "c": 4})) == Xdict({"a": 3, "b": 2, "c": 4}) 206 ``` 207 """ 208 return Xdict.from_list(list(self.items()) + list(other.items())) 209 210 def keys(self) -> Xlist[Y]: 211 """Return an Xlist of the keys of the Xdict.""" 212 return Xlist(self.__data.keys()) 213 214 def values(self) -> Xlist[X]: 215 """Return an Xlist of the values of the Xdict.""" 216 return Xlist(self.__data.values()) 217 218 def items(self) -> Xlist[tuple[Y, X]]: 219 """Return an Xlist of the couples (key, value) of the Xdict.""" 220 return Xlist(self.__data.items()) 221 222 def map[T, U](self, f: F1[[Y, X], tuple[U, T]]) -> "Xdict[U, T]": 223 """Return a new Xdict, after transformation of the couples (key, value) through `f`. 224 225 Transform each couple with `f`, then recreate an Xdict with the result. 226 In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior, 227 meaning this case should be avoided. 228 However, consistency between executions is enforced and the same Xdict will be returned each time. 229 230 ### Usage 231 232 ```python 233 from xfp import Xdict 234 235 assert Xdict({"a": 1, "b": 2}).map(lambda y, x: (f"{y}{y}", x * 10)) == Xdict({"aa": 10, "bb": 20}) 236 237 collisioned = Xdict({"a": 1, "b": 2}).map(lambda _, x: ("c", x * 10)) 238 assert collisioned == Xdict({"c": 20}) or collisioned == Xdict({"c": 10}) # but it will always return the same 239 ``` 240 """ 241 return Xdict.from_list(self.items().map(tupled(f))) 242 243 def map_keys[U](self, f: F1[[Y], U]) -> "Xdict[U, X]": 244 """Return a new Xdict, after transformation of the keys through `f`. 245 246 Transform each key with `f`, then recreate an Xdict with the result. 247 In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior, 248 meaning this case should be avoided. 249 However, consistency between executions is enforced and the same Xdict will be returned each time. 250 251 ### Usage 252 253 ```python 254 from xfp import Xdict 255 256 assert Xdict({"a": 1, "b": 2}).map_keys(lambda y: f"{y}{y}") == Xdict({"aa": 1, "bb": 2}) 257 258 collisioned = Xdict({"a": 1, "b": 2}).map(lambda _: "c") 259 assert collisioned == Xdict({"c": 2}) or collisioned == Xdict({"c": 1}) # but it will always return the same 260 ``` 261 """ 262 return self.map(lambda y, x: (f(y), x)) 263 264 def map_values[T](self, f: F1[[X], T]) -> "Xdict[Y, T]": 265 """Return a new Xdict, after transformation of the values through `f`. 266 267 Transform each value with `f`, then recreate an Xdict with the result. 268 269 ### Usage 270 271 ```python 272 from xfp import Xdict 273 274 assert Xdict({"a": 1, "b": 2}).map_values(lambda x: x * 10) == Xdict({"a": 10, "b": 20}) 275 ``` 276 """ 277 return self.map(lambda y, x: (y, f(x))) 278 279 def filter(self, predicate: F1[[Y, X], bool]) -> "Xdict[Y, X]": 280 """Return a new Xdict, with the couples not matching the predicate deleted. 281 282 Filter the Xdict couples (key, value) using `predicate`. 283 Couples that doesn't match are deleted. 284 285 ### Usage 286 287 ```python 288 from xfp import Xdict 289 290 assert Xdict({"a": "a", "b": "c"}).filter(lambda y, x: y == x) == Xdict({"a": "a"}) 291 ``` 292 """ 293 return self.from_list(self.items().filter(tupled(predicate))) 294 295 def filter_keys(self, predicate: F1[[Y], bool]) -> "Xdict[Y, X]": 296 """Return a new Xdict, with the couples not matching the predicate deleted. 297 298 Filter the Xdict keys using `predicate`. 299 Keys that doesn't match are deleted. 300 301 ### Usage 302 303 ```python 304 from xfp import Xdict 305 306 assert Xdict({"a": 1, "b": 20}).filter(lambda y: y in ["a", "c"]) == Xdict({"a": 1}) 307 ``` 308 """ 309 return self.filter(lambda y, _: predicate(y)) 310 311 def filter_values(self, predicate: F1[[X], bool]) -> "Xdict[Y, X]": 312 """Return a new Xdict, with the couples not matching the predicate deleted. 313 314 Filter the Xdict values using `predicate`. 315 Values that doesn't match are deleted. 316 317 ### Usage 318 319 ```python 320 from xfp import Xdict 321 322 assert Xdict({"a": 1, "b": 20}).filter(lambda x: x < 10) == Xdict({"a": 1}) 323 ``` 324 """ 325 return self.filter(lambda _, x: predicate(x)) 326 327 def foreach(self, statement: F1[[Y, X], Any]) -> None: 328 """Do the 'statement' procedure once for each couple (key, value) of the Xdict. 329 330 ### Usage 331 332 ```python 333 from xfp import Xdict 334 335 input = Xlist({"a": 1, "b": 2}) 336 statement = lambda y, x: print(f"This is an element of the dict : ({y}, {x})") 337 input.foreach(statement) 338 # This is an element of the dict : (a, 1) 339 # This is an element of the dict : (b, 2) 340 ``` 341 """ 342 self.items().foreach(tupled(statement)) 343 344 def foreach_keys(self, statement: F1[[Y], Any]) -> None: 345 """Do the 'statement' procedure once for each key of the Xdict. 346 347 ### Usage 348 349 ```python 350 from xfp import Xdict 351 352 input = Xlist({"a": 1, "b": 2}) 353 statement = lambda y: print(f"This is a key of the dict : {y}") 354 input.foreach_keys(statement) 355 # This is a key of the dict : a 356 # This is a key of the dict : b 357 ``` 358 """ 359 return self.foreach(lambda y, _: statement(y)) 360 361 def foreach_values(self, statement: F1[[X], Any]) -> None: 362 """Do the 'statement' procedure once for each value of the Xdict. 363 364 ### Usage 365 366 ```python 367 from xfp import Xdict 368 369 input = Xlist({"a": 1, "b": 2}) 370 statement = lambda x: print(f"This is a value of the dict : {x}") 371 input.foreach_values(statement) 372 # This is a value of the dict : 1 373 # This is a value of the dict : 2 374 ``` 375 """ 376 return self.foreach(lambda _, x: statement(x))
Abstract base class for generic types.
On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::
class Mapping[KT, VT]:
def __getitem__(self, key: KT) -> VT:
...
# Etc.
On older versions of Python, however, generic classes have to explicitly inherit from Generic.
After a class has been declared to be generic, it can then be used as follows::
def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
try:
return mapping[key]
except KeyError:
return default
Xdict(dic: xfp.xdict.ABCDict[Y, X])
53 def __init__(self, dic: ABCDict[Y, X]) -> None: 54 """Construct an Xdict from a dict-like. 55 56 Dict-like is defined by the existence of the "items" method. 57 """ 58 self.__data = dict(dic.items())
Construct an Xdict from a dict-like.
Dict-like is defined by the existence of the "items" method.
@classmethod
def
from_list(cls, iterable: Iterable[tuple[Y, X]]) -> 'Xdict[Y, X]':
35 @classmethod 36 def from_list(cls, iterable: ABCIterable[tuple[Y, X]]) -> "Xdict[Y, X]": 37 """Return a new Xdict built from an iterable. 38 39 The iterable must contain couples of <key, values>. In case of key duplication 40 in the parameters list, the last associated value is kept. 41 42 ### Usage 43 44 ```python 45 from xfp import Xdict 46 47 xdict = Xdict.from_list([("a", 1), ("b", 2), ("a", 3)]) 48 assert xdict == Xdict({"a": 3, "b": 2}) 49 ``` 50 """ 51 return cls({k: v for k, v in iterable})
Return a new Xdict built from an iterable.
The iterable must contain couples of
Usage
from xfp import Xdict
xdict = Xdict.from_list([("a", 1), ("b", 2), ("a", 3)])
assert xdict == Xdict({"a": 3, "b": 2})
def
get(self, *args) -> X:
116 def get(self, *args) -> X: 117 """Return the value associated with a given key. 118 119 Implementation for the different `get` methods. 120 Should not be called outside of the provided overloaded signatures. 121 122 ### Raises 123 124 - IndexError : if the key is not found in the Xdict and no default is provided 125 - AttributeError : if the method is called with an unspecified set of parameters (signature not found in overload) 126 """ 127 match args: 128 case [y] if y in self.keys(): 129 return cast(X, self.__data.get(y)) 130 case [y]: 131 raise IndexError(f"Key not found in Xdict : {y}") 132 case [y, default]: 133 return self.__data.get(y, default) 134 case _: 135 raise AttributeError("Wrong set of parameters for <get> method")
Return the value associated with a given key.
Implementation for the different get methods.
Should not be called outside of the provided overloaded signatures.
Raises
- IndexError : if the key is not found in the Xdict and no default is provided
- AttributeError : if the method is called with an unspecified set of parameters (signature not found in overload)
137 def get_fr(self, y: Y, /) -> Xresult[IndexError, X]: 138 """Return the value associated with a given key. 139 140 Wrap the result in an Xtry to catch potential errors. 141 142 ### Returns 143 144 - Xtry.Success : an Xresult containing the value if the key is found 145 - Xtry.Failure : with an IndexError if the key is not found 146 147 ### Usage 148 149 ```python 150 from xfp import Xdict, Xtry 151 152 match Xdict({"a": 1}).get_fr("b"): 153 case Xtry.Success(value): 154 print(f"we found {value} in Xdict") 155 case Xtry.Failure(e): 156 print(f"Error: {e}") 157 ``` 158 """ 159 return cast(Xresult[IndexError, X], Xtry.from_unsafe(lambda: self.get(y)))
Return the value associated with a given key.
Wrap the result in an Xtry to catch potential errors.
Returns
- Xtry.Success : an Xresult containing the value if the key is found
- Xtry.Failure : with an IndexError if the key is not found
Usage
from xfp import Xdict, Xtry
match Xdict({"a": 1}).get_fr("b"):
case Xtry.Success(value):
print(f"we found {value} in Xdict")
case Xtry.Failure(e):
print(f"Error: {e}")
def
updated(self, key: Y, value: T) -> 'Xdict[Y, X | T]':
161 def updated[T](self, key: Y, value: T) -> "Xdict[Y, X | T]": 162 """Return a new Xdict, with an updated couple (key: Y, value: E). 163 164 Upsert a new `value` at `key`. 165 166 ### Usage 167 168 ```python 169 from xfp import Xdict 170 171 assert Xdict({"a": 1}).updated("b", 2) == Xdict({"a": 1, "b": 2}) 172 assert Xdict({"a": 1}).updated("a", 2) == Xdict({"a": 2}) 173 ``` 174 """ 175 return self.union(Xdict({key: value}))
Return a new Xdict, with an updated couple (key: Y, value: E).
Upsert a new value at key.
Usage
from xfp import Xdict
assert Xdict({"a": 1}).updated("b", 2) == Xdict({"a": 1, "b": 2})
assert Xdict({"a": 1}).updated("a", 2) == Xdict({"a": 2})
def
removed(self, key: Y) -> 'Xdict[Y, X]':
177 def removed(self, key: Y) -> "Xdict[Y, X]": 178 """Return a new Xdict, with the given key deleted. 179 180 Filter the provided key if found. 181 No error is raised if the key doesn't exist. 182 183 ### Usage 184 185 ```python 186 from xfp import Xdict 187 188 assert Xdict({"a": 1, "b": 2}).removed("b") == Xdict({"a": 1}) 189 assert Xdict({"a": 1}).removed("b") == Xdict({"a": 1}) 190 ``` 191 """ 192 return self.filter(lambda y, _: y is not key)
Return a new Xdict, with the given key deleted.
Filter the provided key if found. No error is raised if the key doesn't exist.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 2}).removed("b") == Xdict({"a": 1})
assert Xdict({"a": 1}).removed("b") == Xdict({"a": 1})
def
union(self, other: xfp.xdict.ABCDict[U, T]) -> 'Xdict[Y | U, X | T]':
194 def union[T, U](self, other: ABCDict[U, T]) -> "Xdict[Y | U, X | T]": 195 """Return a new Xdict, being the merge of self and a given one. 196 197 Works as if multiple updateds are done successively. 198 It means if a key is present in both Xdict, the `other` Xdict has priority. 199 200 ### Usage 201 202 ```python 203 from xfp import Xdict 204 205 assert Xdict({"a": 1, "b": 2}).union(Xdict({"a": 3, "c": 4})) == Xdict({"a": 3, "b": 2, "c": 4}) 206 ``` 207 """ 208 return Xdict.from_list(list(self.items()) + list(other.items()))
Return a new Xdict, being the merge of self and a given one.
Works as if multiple updateds are done successively.
It means if a key is present in both Xdict, the other Xdict has priority.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 2}).union(Xdict({"a": 3, "c": 4})) == Xdict({"a": 3, "b": 2, "c": 4})
210 def keys(self) -> Xlist[Y]: 211 """Return an Xlist of the keys of the Xdict.""" 212 return Xlist(self.__data.keys())
Return an Xlist of the keys of the Xdict.
214 def values(self) -> Xlist[X]: 215 """Return an Xlist of the values of the Xdict.""" 216 return Xlist(self.__data.values())
Return an Xlist of the values of the Xdict.
218 def items(self) -> Xlist[tuple[Y, X]]: 219 """Return an Xlist of the couples (key, value) of the Xdict.""" 220 return Xlist(self.__data.items())
Return an Xlist of the couples (key, value) of the Xdict.
def
map(self, f: F1[[Y, X], tuple[U, T]]) -> 'Xdict[U, T]':
222 def map[T, U](self, f: F1[[Y, X], tuple[U, T]]) -> "Xdict[U, T]": 223 """Return a new Xdict, after transformation of the couples (key, value) through `f`. 224 225 Transform each couple with `f`, then recreate an Xdict with the result. 226 In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior, 227 meaning this case should be avoided. 228 However, consistency between executions is enforced and the same Xdict will be returned each time. 229 230 ### Usage 231 232 ```python 233 from xfp import Xdict 234 235 assert Xdict({"a": 1, "b": 2}).map(lambda y, x: (f"{y}{y}", x * 10)) == Xdict({"aa": 10, "bb": 20}) 236 237 collisioned = Xdict({"a": 1, "b": 2}).map(lambda _, x: ("c", x * 10)) 238 assert collisioned == Xdict({"c": 20}) or collisioned == Xdict({"c": 10}) # but it will always return the same 239 ``` 240 """ 241 return Xdict.from_list(self.items().map(tupled(f)))
Return a new Xdict, after transformation of the couples (key, value) through f.
Transform each couple with f, then recreate an Xdict with the result.
In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior,
meaning this case should be avoided.
However, consistency between executions is enforced and the same Xdict will be returned each time.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 2}).map(lambda y, x: (f"{y}{y}", x * 10)) == Xdict({"aa": 10, "bb": 20})
collisioned = Xdict({"a": 1, "b": 2}).map(lambda _, x: ("c", x * 10))
assert collisioned == Xdict({"c": 20}) or collisioned == Xdict({"c": 10}) # but it will always return the same
def
map_keys(self, f: F1[[Y], U]) -> 'Xdict[U, X]':
243 def map_keys[U](self, f: F1[[Y], U]) -> "Xdict[U, X]": 244 """Return a new Xdict, after transformation of the keys through `f`. 245 246 Transform each key with `f`, then recreate an Xdict with the result. 247 In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior, 248 meaning this case should be avoided. 249 However, consistency between executions is enforced and the same Xdict will be returned each time. 250 251 ### Usage 252 253 ```python 254 from xfp import Xdict 255 256 assert Xdict({"a": 1, "b": 2}).map_keys(lambda y: f"{y}{y}") == Xdict({"aa": 1, "bb": 2}) 257 258 collisioned = Xdict({"a": 1, "b": 2}).map(lambda _: "c") 259 assert collisioned == Xdict({"c": 2}) or collisioned == Xdict({"c": 1}) # but it will always return the same 260 ``` 261 """ 262 return self.map(lambda y, x: (f(y), x))
Return a new Xdict, after transformation of the keys through f.
Transform each key with f, then recreate an Xdict with the result.
In case of conflicts in the resulted key set, which value is associated with each key is an unenforced behavior,
meaning this case should be avoided.
However, consistency between executions is enforced and the same Xdict will be returned each time.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 2}).map_keys(lambda y: f"{y}{y}") == Xdict({"aa": 1, "bb": 2})
collisioned = Xdict({"a": 1, "b": 2}).map(lambda _: "c")
assert collisioned == Xdict({"c": 2}) or collisioned == Xdict({"c": 1}) # but it will always return the same
def
map_values(self, f: F1[[X], T]) -> 'Xdict[Y, T]':
264 def map_values[T](self, f: F1[[X], T]) -> "Xdict[Y, T]": 265 """Return a new Xdict, after transformation of the values through `f`. 266 267 Transform each value with `f`, then recreate an Xdict with the result. 268 269 ### Usage 270 271 ```python 272 from xfp import Xdict 273 274 assert Xdict({"a": 1, "b": 2}).map_values(lambda x: x * 10) == Xdict({"a": 10, "b": 20}) 275 ``` 276 """ 277 return self.map(lambda y, x: (y, f(x)))
Return a new Xdict, after transformation of the values through f.
Transform each value with f, then recreate an Xdict with the result.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 2}).map_values(lambda x: x * 10) == Xdict({"a": 10, "b": 20})
def
filter(self, predicate: F1[[Y, X], bool]) -> 'Xdict[Y, X]':
279 def filter(self, predicate: F1[[Y, X], bool]) -> "Xdict[Y, X]": 280 """Return a new Xdict, with the couples not matching the predicate deleted. 281 282 Filter the Xdict couples (key, value) using `predicate`. 283 Couples that doesn't match are deleted. 284 285 ### Usage 286 287 ```python 288 from xfp import Xdict 289 290 assert Xdict({"a": "a", "b": "c"}).filter(lambda y, x: y == x) == Xdict({"a": "a"}) 291 ``` 292 """ 293 return self.from_list(self.items().filter(tupled(predicate)))
Return a new Xdict, with the couples not matching the predicate deleted.
Filter the Xdict couples (key, value) using predicate.
Couples that doesn't match are deleted.
Usage
from xfp import Xdict
assert Xdict({"a": "a", "b": "c"}).filter(lambda y, x: y == x) == Xdict({"a": "a"})
def
filter_keys(self, predicate: F1[[Y], bool]) -> 'Xdict[Y, X]':
295 def filter_keys(self, predicate: F1[[Y], bool]) -> "Xdict[Y, X]": 296 """Return a new Xdict, with the couples not matching the predicate deleted. 297 298 Filter the Xdict keys using `predicate`. 299 Keys that doesn't match are deleted. 300 301 ### Usage 302 303 ```python 304 from xfp import Xdict 305 306 assert Xdict({"a": 1, "b": 20}).filter(lambda y: y in ["a", "c"]) == Xdict({"a": 1}) 307 ``` 308 """ 309 return self.filter(lambda y, _: predicate(y))
Return a new Xdict, with the couples not matching the predicate deleted.
Filter the Xdict keys using predicate.
Keys that doesn't match are deleted.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 20}).filter(lambda y: y in ["a", "c"]) == Xdict({"a": 1})
def
filter_values(self, predicate: F1[[X], bool]) -> 'Xdict[Y, X]':
311 def filter_values(self, predicate: F1[[X], bool]) -> "Xdict[Y, X]": 312 """Return a new Xdict, with the couples not matching the predicate deleted. 313 314 Filter the Xdict values using `predicate`. 315 Values that doesn't match are deleted. 316 317 ### Usage 318 319 ```python 320 from xfp import Xdict 321 322 assert Xdict({"a": 1, "b": 20}).filter(lambda x: x < 10) == Xdict({"a": 1}) 323 ``` 324 """ 325 return self.filter(lambda _, x: predicate(x))
Return a new Xdict, with the couples not matching the predicate deleted.
Filter the Xdict values using predicate.
Values that doesn't match are deleted.
Usage
from xfp import Xdict
assert Xdict({"a": 1, "b": 20}).filter(lambda x: x < 10) == Xdict({"a": 1})
def
foreach(self, statement: F1[[Y, X], typing.Any]) -> None:
327 def foreach(self, statement: F1[[Y, X], Any]) -> None: 328 """Do the 'statement' procedure once for each couple (key, value) of the Xdict. 329 330 ### Usage 331 332 ```python 333 from xfp import Xdict 334 335 input = Xlist({"a": 1, "b": 2}) 336 statement = lambda y, x: print(f"This is an element of the dict : ({y}, {x})") 337 input.foreach(statement) 338 # This is an element of the dict : (a, 1) 339 # This is an element of the dict : (b, 2) 340 ``` 341 """ 342 self.items().foreach(tupled(statement))
Do the 'statement' procedure once for each couple (key, value) of the Xdict.
Usage
from xfp import Xdict
input = Xlist({"a": 1, "b": 2})
statement = lambda y, x: print(f"This is an element of the dict : ({y}, {x})")
input.foreach(statement)
# This is an element of the dict : (a, 1)
# This is an element of the dict : (b, 2)
def
foreach_keys(self, statement: F1[[Y], typing.Any]) -> None:
344 def foreach_keys(self, statement: F1[[Y], Any]) -> None: 345 """Do the 'statement' procedure once for each key of the Xdict. 346 347 ### Usage 348 349 ```python 350 from xfp import Xdict 351 352 input = Xlist({"a": 1, "b": 2}) 353 statement = lambda y: print(f"This is a key of the dict : {y}") 354 input.foreach_keys(statement) 355 # This is a key of the dict : a 356 # This is a key of the dict : b 357 ``` 358 """ 359 return self.foreach(lambda y, _: statement(y))
Do the 'statement' procedure once for each key of the Xdict.
Usage
from xfp import Xdict
input = Xlist({"a": 1, "b": 2})
statement = lambda y: print(f"This is a key of the dict : {y}")
input.foreach_keys(statement)
# This is a key of the dict : a
# This is a key of the dict : b
def
foreach_values(self, statement: F1[[X], typing.Any]) -> None:
361 def foreach_values(self, statement: F1[[X], Any]) -> None: 362 """Do the 'statement' procedure once for each value of the Xdict. 363 364 ### Usage 365 366 ```python 367 from xfp import Xdict 368 369 input = Xlist({"a": 1, "b": 2}) 370 statement = lambda x: print(f"This is a value of the dict : {x}") 371 input.foreach_values(statement) 372 # This is a value of the dict : 1 373 # This is a value of the dict : 2 374 ``` 375 """ 376 return self.foreach(lambda _, x: statement(x))
Do the 'statement' procedure once for each value of the Xdict.
Usage
from xfp import Xdict
input = Xlist({"a": 1, "b": 2})
statement = lambda x: print(f"This is a value of the dict : {x}")
input.foreach_values(statement)
# This is a value of the dict : 1
# This is a value of the dict : 2