Coverage for src/pyfplib/

1# SPDX-FileCopyrightText: 2026-present Comet11x

2# SPDX-License-Identifier: MIT

3"""This module provides Option[T] and Result[T, E]"""

5from collections.abc import Callable

6from typing import TypeVar, cast, overload

8from pyfplib.errors import ExpectedError, UnwrapError

10Ts = TypeVar("Ts")

13class Result[T, E]:

14 """

15 Result<T, E> monad for Python.

17 Represents a value that is either Some(T) containing data,

18 or Nothing() representing absence of value.

20 Provides safe null-handling through method chaining

21 instead of explicit None checks.

22 """

24 # Lets positional patterns bind to `value`, e.g. `case Ok(v)`.

25 __match_args__ = ("value",)

27 def __init__(self, *, value: T | None = None, error: E | None = None):

28 # A result is Ok exactly when no error was supplied; the two states share one slot.

29 self.__is_ok = error is None

30 self.__value: T | E = cast(T, value) if self.__is_ok else cast(E, error)

32 @property

33 def value(self) -> T | E:

34 """

35 Returns a contained value.

37 Returns:

38 T | E: The Ok value or the Err error.

39 """

40 return self.__value

42 def is_ok(self) -> bool:

43 """

44 Returns True if the result is Ok[T, E].

46 Returns:

47 bool: True when this result holds an Ok value.

48 """

49 return self.__is_ok

51 def is_err(self) -> bool:

52 """

53 Returns True if the result is Err[T, E].

55 Returns:

56 bool: True when this result holds an Err value.

57 """

58 return not self.__is_ok

60 def __bool__(self) -> bool:

61 """

62 Boolean conversion - True if Ok, False if Err.

64 Returns:

65 bool: The truthiness of this result.

66 """

67 return self.is_ok()

69 def __repr__(self) -> str:

70 """

71 Returns a readable representation of this result.

73 Returns:

74 str: ``Ok(value)`` or ``Err(error)``.

75 """

76 return f"Ok({self.__value!r})" if self.is_ok() else f"Err({self.__value!r})"

78 def inspect(self, fn: Callable[[T], None]) -> "Result[T, E]":

79 """

80 Applies the given function fn to a contained value if the result is Ok[T, E].

82 Args:

83 fn: Callback invoked with the Ok value.

85 Returns:

86 Result[T, E]: This result, for method chaining.

87 """

88 if self.is_ok():

89 fn(cast(T, self.__value))

90 return self

92 def inspect_err(self, fn: Callable[[E], None]) -> "Result[T, E]":

93 """

94 Applies the given function fn to a contained value if the result is Err[T, E].

96 Args:

97 fn: Callback invoked with the Err value.

99 Returns:

100 Result[T, E]: This result, for method chaining.

101 """

102 if self.is_err():

103 fn(cast(E, self.__value))

104 return self

105

106 def ok(self) -> "Option[T]":

107 """

108 Returns Some[T] if this result is Ok[T, E], otherwise Nothing[T].

109

110 Returns:

111 Option[T]: The Ok value as an Option.

112 """

113 return Some(cast(T, self.__value)) if self.is_ok() else Nothing()

114

115 def err(self) -> "Option[E]":

116 """

117 Returns Some[E] if this result is Err[E], otherwise Nothing[E].

118

119 Returns:

120 Option[E]: The Err value as an Option.

121 """

122 return Some(cast(E, self.__value)) if self.is_err() else Nothing()

123

124 def unwrap(self) -> T:

125 """

126 Extracts a contained value, raising UnwrapError if Err.

127

128 Returns:

129 T: The contained Ok value.

130

131 Raises:

132 UnwrapError: When attempting to unwrap Err

133 """

134 if self.is_err():

135 msg = "called `Result.unwrap()` on a `Err` value"

136 raise UnwrapError(msg)

137 value: T = cast(T, self.__value)

138 return value

139

140 def unwrap_err(self) -> E:

141 """

142 Extracts a contained error, raising UnwrapError if Ok.

143

144 Returns:

145 E: The contained Err value.

146

147 Raises:

148 UnwrapError: When attempting to unwrap Ok

149 """

150 if self.is_ok():

151 msg = "called `Result.unwrap_err()` on a `Ok` value"

152 raise UnwrapError(msg)

153 return cast(E, self.__value)

154

155 def expect(self, message: str) -> T:

156 """

157 Returns a contained value or raises ExpectedError.

158

159 Args:

160 message: Error message used when this result is Err.

161

162 Returns:

163 T: The contained Ok value.

164

165 Raises:

166 ExpectedError: When this result is Err.

167 """

168 if self.is_err():

169 raise ExpectedError(message)

170 return cast(T, self.__value)

171

172 def expect_err(self, message: str) -> E:

173 """

174 Returns a contained error or raises ExpectedError.

175

176 Args:

177 message: Error message used when this result is Ok.

178

179 Returns:

180 E: The contained Err value.

181

182 Raises:

183 ExpectedError: When this result is Ok.

184 """

185 if self.is_ok():

186 raise ExpectedError(message)

187 return cast(E, self.__value)

188

189 def unwrap_or(self, default: T) -> T:

190 """

191 Returns a contained value or provided default if the result is Err[T, E].

192

193 Args:

194 default: Value returned when this result is Err.

195

196 Returns:

197 T: The Ok value or `default`.

198 """

199 return cast(T, self.__value) if self.is_ok() else default

200

201 def unwrap_or_else(self, fn: Callable[[E], T]) -> T:

202 """

203 Returns a contained value or computes a fallback from the Err value.

204

205 Args:

206 fn: Callback receiving the Err value and returning a fallback.

207

208 Returns:

209 T: The Ok value or the value produced by `fn`.

210 """

211 return cast(T, self.__value) if self.is_ok() else fn(cast(E, self.__value))

212

213 def unwrap_err_or(self, default: E) -> E:

214 """

215 Returns a contained error or provided default if the result is Ok[T, E].

216

217 Args:

218 default: Value returned when this result is Ok.

219

220 Returns:

221 E: The Err value or `default`.

222 """

223 return cast(E, self.__value) if self.is_err() else default

224

225 def map[U](self, fn: Callable[[T], U]) -> "Result[U, E]":

226 """

227 Maps a Result[T, E] to Result[U, E].

228

229 Args:

230 fn: Callback applied to the Ok value.

231

232 Returns:

233 Result[U, E]: A new Ok with the mapped value, or the original Err.

234 """

235 if self.is_ok():

236 return Ok(fn(self.unwrap()))

237 else:

238 return Err(self.unwrap_err())

239

240 def map_or[U](self, default: T, fn: Callable[[T], U]) -> "Result[U, E]":

241 """

242 Maps a Result[T, E] to Result[U, E], using a default for Err values.

243

244 Args:

245 default: Value used when this result is Err.

246 fn: Callback applied to the Ok value or `default`.

247

248 Returns:

249 Result[U, E]: An Ok wrapping the mapped value.

250 """

251 value: T = cast(T, self.__value) if self.is_ok() else default

252 return cast(Result[U, E], Ok(fn(value)))

253

254 def map_or_else[U](self, default: Callable[[E], T], fn: Callable[[T], U]) -> U:

255 """

256 Maps Result[T, E] to U by applying the given function fn to a contained Ok value.

257 If the result is Err, it applies the given fallback function default to the Err value.

258

259 Args:

260 default: Callback receiving the Err value and returning a fallback.

261 fn: Callback applied to the Ok value or the fallback result.

262

263 Returns:

264 U: The mapped value.

265 """

266 value: T = cast(T, self.__value) if self.is_ok() else default(self.unwrap_err())

267 return fn(value)

268

269 def map_err[U](self, fn: Callable[[E], U]) -> "Result[T, U]":

270 """

271 Maps the Err value of a Result[T, E] to Result[T, U].

272

273 Args:

274 fn: Callback applied to the Err value.

275

276 Returns:

277 Result[T, U]: The original Ok, or an Err with the mapped error.

278 """

279 # Transform only the error branch, leaving Ok values untouched.

280 if self.is_err():

281 res_err: Result[T, U] = Err(fn(cast(E, self.__value)))

282 return res_err

283 else:

284 res_ok: Result[T, U] = Ok(cast(T, self.__value))

285 return res_ok

286

287 @overload

288 def flatten(self: "Result[Result[T, E], E]") -> "Result[T, E]": ...

289

290 @overload

291 def flatten(self: "Result[T, E]") -> "Result[T, E]": ...

292

293 def flatten(self):

294 """

295 Collapses one level of nesting.

296

297 Returns:

298 Result[T, E]: The inner result when nested, otherwise this result.

299 """

300 # Collapse exactly one level of nesting; non-nested results are returned as-is.

301 return self.__value if isinstance(self.__value, Result) else self

302

303 @overload

304 def flatten_all(self: "Result[Result[T, E], E]") -> "Result[T, E]": ...

305

306 @overload

307 def flatten_all(self: "Result[T, E]") -> "Result[T, E]": ...

308

309 def flatten_all(self) -> "Result":

310 """

311 Collapses every level of nesting.

312

313 Returns:

314 Result[T, E]: The innermost result.

315 """

316 # Walk down nested results until reaching a value that is not a Result.

317 value: Result | object = self

318 while isinstance(value.__value, Result):

319 value: Result = value.__value

320 return value

321

322 def __and__(self, other: "Result[T, E] | Callable[[T], Result[T, E]]") -> "Result[T, E]":

323 """

324 Chains results with `and` semantics.

325

326 Args:

327 other: A result or a callback returning one, used when this result is Ok.

328

329 Returns:

330 Result[T, E]: This Err, or the result produced from `other`.

331 """

332 # `and` semantics: an Err short-circuits, while an Ok defers to `other`.

333 if self.is_err():

334 return self

335 else:

336 if isinstance(other, Callable):

337 fn = cast(Callable[[T], "Result[T, E]"], other)

338 return fn(cast(T, self.__value))

339 else:

340 return cast(Result[T, E], other)

341

342 def and_then(self, fn: Callable[[T], "Result[T, E]"]) -> "Result[T, E]":

343 """

344 Runs the next step only when this result is Ok.

345

346 Args:

347 fn: Callback receiving the Ok value and returning a new result.

348

349 Returns:

350 Result[T, E]: The callback result, or this Err unchanged.

351 """

352 # Monadic bind: run the next step only when this result is Ok.

353 return fn(cast(T, self.__value)) if self.is_ok() and callable(fn) else self

354

355 def __or__(self, other: "Result[T, E] | Callable[[E], Result[T, E]]") -> "Result[T, E]":

356 """

357 Chains results with `or` semantics.

358

359 Args:

360 other: A result or a callback returning one, used when this result is Err.

361

362 Returns:

363 Result[T, E]: This Ok, or the result produced from `other`.

364 """

365 # `or` semantics: an Ok short-circuits, while an Err defers to `other`.

366 if self.is_ok():

367 res = self

368 else:

369 if isinstance(other, Callable):

370 fn = cast(Callable[[E], "Result[T, E]"], other)

371 res = fn(cast(E, self.__value))

372 else:

373 res = cast(Result[T, E], other)

374 return res

375

376 def or_else(self, fn: Callable[[E], "Result[T, E]"]) -> "Result[T, E]":

377 """

378 Runs a recovery step only when this result is Err.

379

380 Args:

381 fn: Callback receiving the Err value and returning a new result.

382

383 Returns:

384 Result[T, E]: The callback result, or this Ok unchanged.

385 """

386 # Recovery step: invoked only for an Err value.

387 return fn(cast(E, self.__value)) if self.is_err() and callable(fn) else self

388

389 @staticmethod

390 def try_call[R](fn: Callable[..., R], *args, **kwargs) -> "Result[R, Exception]":

391 """

392 Tries to call the given function fn and returns a result of the execution.

393

394 Args:

395 fn: Callable to invoke.

396 *args: Positional arguments forwarded to `fn`.

397 **kwargs: Keyword arguments forwarded to `fn`.

398

399 Returns:

400 Result[R, Exception]: Ok with the return value, or Err with the raised exception.

401 """

402 # Turn any raised exception into an Err value instead of letting it propagate.

403 try:

404 ret: R = fn(*args, **kwargs)

405 return Ok(ret)

406 except Exception as err: # noqa: BLE001

407 return Err(err)

408

409

410class Ok[T, E](Result[T, E]):

411 """Result constructor."""

412

413 def __init__(self, value: T):

414 super().__init__(value=value)

415

416

417class Err[T, E](Result[T, E]):

418 """Result constructor."""

419

420 def __init__(self, error: E):

421 super().__init__(error=error)

422

423

424class Option[T]:

425 """

426 Option<T> monad for Python.

427

428 Represents a value that is either Some(T) containing data,

429 or Nothing() representing absence of value.

430

431 Provides safe null-handling through method chaining

432 instead of explicit None checks.

433 """

434

435 # Support for structural pattern matching (Python 3.10+)

436 __match_args__ = ("value",)

437

438 def __init__(self, value: T | None):

439 """

440 Internal constructor - use Some(value) or Nothing() instead.

441

442 Args:

443 value: The contained value or None for Nothing

444 """

445 self.__value = value

446

447 @property

448 def value(self) -> T | None:

449 """

450 Returns the raw contained value (use with caution - may be None).

451

452 Returns:

453 T | None: The contained value, or None for Nothing.

454 """

455 return self.__value

456

457 def is_some(self) -> bool:

458 """

459 Returns True if this Option contains a value (Some[T]).

460

461 Returns:

462 bool: True when a value is present.

463 """

464 return self.__value is not None

465

466 def is_none(self) -> bool:

467 """

468 Returns True if this Option contains no value (Nothing).

469

470 Returns:

471 bool: True when no value is present.

472 """

473 return self.__value is None

474

475 def __bool__(self) -> bool:

476 """

477 Boolean conversion - True if Some[T], False if Nothing.

478

479 Returns:

480 bool: The truthiness of this option.

481 """

482 return self.is_some()

483

484 def __repr__(self) -> str:

485 """

486 Returns a readable representation of this option.

487

488 Returns:

489 str: ``Some(value)`` or ``Nothing()``.

490 """

491 return "Nothing()" if self.is_none() else f"Some({self.__value!r})"

492

493 def inspect(self, fn: Callable[[T], None]) -> "Option[T]":

494 """

495 Executes callback if Some[T], otherwise does nothing.

496

497 Args:

498 fn: callback to execute with contained value

499

500 Returns:

501 Option[T]: This option, for method chaining.

502 """

503 if self.__value is not None:

504 fn(self.__value)

505 return self

506

507 def inspect_none(self, fn: Callable[[], None]) -> "Option[T]":

508 """

509 Executes callback if Nothing, otherwise does nothing.

510

511 Args:

512 fn: callback to execute when no value present

513

514 Returns:

515 Option[T]: This option, for method chaining.

516 """

517 if self.__value is None:

518 fn()

519 return self

520

521 def unwrap(self) -> T:

522 """

523 Extracts the contained value, raising UnwrapError if Nothing.

524

525 Returns:

526 T: The contained value.

527

528 Raises:

529 UnwrapError: When attempting to unwrap Nothing

530 """

531 if self.__value is None:

532 msg = "called `Option.unwrap()` on a `Nothing` value"

533 raise UnwrapError(msg)

534 return self.__value

535

536 def expect(self, message: str) -> T:

537 """

538 Extracts the contained value, raising ExpectedError if Nothing with the given message.

539

540 Args:

541 message: an error message

542

543 Returns:

544 T: The contained value.

545

546 Raises:

547 ExpectedError: When attempting to unwrap Nothing

548 """

549 if self.__value is None:

550 raise ExpectedError(message)

551 return self.__value

552

553 def unwrap_or(self, default: T) -> T:

554 """

555 Returns contained value or a provided default value if it is Nothing.

556

557 Args:

558 default: a value returned if the option is empty

559

560 Returns:

561 T: The contained value or `default`.

562 """

563 return default if self.is_none() else cast(T, self.__value)

564

565 def unwrap_or_else(self, fn: Callable[[], T]) -> T:

566 """

567 Returns the contained Some value or computes it from a closure.

568

569 Args:

570 fn: Callback producing a fallback value when this option is empty.

571

572 Returns:

573 T: The contained value or the value produced by `fn`.

574 """

575 return fn() if self.is_none() else cast(T, self.__value)

576

577 def map[U](self, fn: Callable[[T], U]) -> "Option[U]":

578 """

579 Maps the contained value with a function.

580

581 Args:

582 fn: a pure function mapping T to U

583

584 Returns:

585 Option[U]: Some with the mapped value, or Nothing.

586 """

587 if self.is_some():

588 return Some(fn(self.unwrap()))

589 else:

590 return Nothing()

591

592 def map_or[U](self, default: U, fn: Callable[[T], U]) -> "Option[U]":

593 """

594 Maps the contained value, using a default when this option is empty.

595

596 Args:

597 default: Value used when this option is Nothing.

598 fn: a pure function mapping T to U

599

600 Returns:

601 Option[U]: An Option wrapping the mapped value or `default`.

602 """

603 value = default if self.is_none() else fn(cast(T, self.__value))

604 return Some(value)

605

606 def map_or_else[U](self, default: Callable[[], U], fn: Callable[[T], U]) -> U:

607 """

608 Maps the contained value or computes a fallback when empty.

609

610 Args:

611 default: Callback producing a fallback when this option is Nothing.

612 fn: a pure function mapping T to U

613

614 Returns:

615 U: The mapped value or the fallback.

616 """

617 return default() if self.is_none() else fn(cast(T, self.__value))

618

619 @overload

620 def flatten(self: "Option[Option[T]]") -> "Option[T]": ...

621

622 @overload

623 def flatten(self: "Option[T]") -> "Option[T]": ...

624

625 def flatten(self):

626 """

627 Collapses one level of nesting.

628

629 Returns:

630 Option[T]: The inner option when nested, otherwise this option.

631 """

632 return self.__value if isinstance(self.__value, Option) else self

633

634 @overload

635 def flatten_all(self: "Option[Option[T]]") -> "Option[T]": ...

636

637 @overload

638 def flatten_all(self: "Option[T]") -> "Option[T]": ...

639

640 def flatten_all(self) -> "Option[T]":

641 """

642 Collapses every level of nesting.

643

644 Returns:

645 Option[T]: The innermost option.

646 """

647 value: Option[T] | object = self

648 while isinstance(value.__value, Option):

649 value = value.__value

650 return value

651

652 def ok_or[E](self, err: E) -> Result[T, E]:

653 """

654 Converts this option into a Result.

655

656 Args:

657 err: Error used when this option is Nothing.

658

659 Returns:

660 Result[T, E]: Ok with the contained value, or Err with `err`.

661 """

662 if self.is_none():

663 res: Result[T, E] = Err(err)

664 else:

665 res: Result[T, E] = Ok(self.unwrap())

666 return res

667

668 def ok_or_else[E](self, fn: Callable[[], E]) -> Result[T, E]:

669 """

670 Converts this option into a Result, computing the error when empty.

671

672 Args:

673 fn: Callback producing the error when this option is Nothing.

674

675 Returns:

676 Result[T, E]: Ok with the contained value, or Err produced by `fn`.

677 """

678 if self.is_none():

679 res: Result[T, E] = Err(fn())

680 else:

681 res: Result[T, E] = Ok(self.unwrap())

682 return res

683

684 def __eq__(self, other: object) -> bool:

685 """

686 Equality comparison based on contained values.

687

688 Args:

689 other: Object to compare against.

690

691 Returns:

692 bool: True when both are Options holding equal values.

693 """

694 ret: bool = False

695 if isinstance(other, Option):

696 ret = self.__value == other.value

697 return ret

698

699 def __hash__(self):

700 """

701 Hashes the contained value.

702

703 Returns:

704 int: The hash of the contained value.

705 """

706 return hash(self.__value)

707

708 def __and__[U](self, other: "Option[U] | Callable[[], Option[U]]") -> "Option[U]":

709 """

710 Chains options with `and` semantics.

711

712 Args:

713 other: An option or a callback returning one, used when this option is Some.

714

715 Returns:

716 Option[U]: Nothing, or the option produced from `other`.

717 """

718 # `and` semantics: Nothing short-circuits, Some defers to `other`.

719 if self.is_some():

720 return cast(Callable[[], Option[U]], other)() if isinstance(other, Callable) else cast(Option[U], other)

721 else:

722 return Nothing()

723

724 def and_then[U](self, fn: Callable[[T], "Option[U]"]) -> "Option[U]":

725 """

726 Runs the next step only when a value is present.

727

728 Args:

729 fn: Callback receiving the contained value and returning a new option.

730

731 Returns:

732 Option[U]: The callback result, or Nothing.

733 """

734 # Monadic bind: run the next step only when a value is present.

735 if self.is_some() and callable(fn):

736 return fn(cast(T, self.__value))

737 else:

738 return Nothing()

739

740 def __or__(self, other: "Option[T] | Callable[[], Option[T]]") -> "Option[T]":

741 """

742 Chains options with `or` semantics.

743

744 Args:

745 other: An option or a callback returning one, used when this option is Nothing.

746

747 Returns:

748 Option[T]: This option, or the option produced from `other`.

749 """

750 # `or` semantics: Some short-circuits, Nothing defers to `other`.

751 if self.is_some():

752 return self

753 else:

754 return cast(Callable[[], Option[T]], other)() if isinstance(other, Callable) else other

755

756 def or_else(self, fn: Callable[[], "Option[T]"]) -> "Option[T]":

757 """

758 Runs a recovery step only when this option is empty.

759

760 Args:

761 fn: Callback returning a fallback option when this option is Nothing.

762

763 Returns:

764 Option[T]: This option, or the callback result.

765 """

766 # Recovery step: invoked only when this option is empty.

767 return fn() if self.is_none() and callable(fn) else self

768

769 def xor(self, other: "Option[T]") -> "Option[T]":

770 """

771 Returns a value only when exactly one of the two options is Some.

772

773 Args:

774 other: Option to combine with.

775

776 Returns:

777 Option[T]: The sole Some value, or Nothing.

778 """

779 # Return a value only when exactly one of the two options is Some.

780 if self.is_some() and other.is_none():

781 return self

782 elif self.is_none() and other.is_some():

783 return other

784 else:

785 return Nothing()

786

787 def push(self, value: T) -> T:

788 """

789 Stores a new value and returns the previously held one.

790

791 Args:

792 value: New value to store.

793

794 Returns:

795 T: The previously held value, or `value` when this option was empty.

796 """

797 # Return the previously held value (or the new one when empty), then store `value`.

798 ret = cast(T, self.__value) if self.is_some() else value

799 self.__value = value

800 return ret

801

802 @staticmethod

803 def from_optional(value: T | None = None) -> "Option[T]":

804 """

805 Creates Option[T] from Optional[T].

806

807 Args:

808 value: Value to wrap; None produces Nothing.

809

810 Returns:

811 Option[T]: Some(value) or Nothing.

812 """

813 return Nothing() if value is None else Some(value)

814

815 def zip[U](self, other: "Option[U]") -> "Option[tuple[T, U]]":

816 """

817 Combines two options into a tuple when both contain a value.

818

819 Args:

820 other: Option to combine with.

821

822 Returns:

823 Option[tuple[T, U]]: A tuple of both values, or Nothing.

824 """

825 # Combine two options only when both contain a value.

826 if self.is_some() and other.is_some():

827 return Some((cast(T, self.__value), cast(U, other.__value)))

828 else:

829 return Nothing()

830

831 def zip_with[U, R](self, other: "Option[U]", ctr: Callable[[T, U], R]) -> "Option[R]":

832 """

833 Combines two options with a callback when both contain a value.

834

835 Args:

836 other: Option to combine with.

837 ctr: Callback receiving both contained values.

838

839 Returns:

840 Option[R]: The callback result, or Nothing.

841 """

842 # Combine two present values with `ctr`, otherwise return Nothing.

843 if self.is_some() and other.is_some():

844 return Some(ctr(cast(T, self.__value), cast(U, other.__value)))

845 else:

846 return Nothing()

847

848

849class Some[T](Option[T]):

850 """

851 Some extends Option[T].

852 It is used to create Option[T] containing a value

853

854 Usage:

855 Some(123)

856 Some("hello")

857 """

858

859 def __init__(self, value: T):

860 super().__init__(value)

861

862

863class Nothing[T](Option[T]):

864 """

865 Nothing extends Option[T].

866 It is used to create empty Option[T]

867

868 Usage:

869 Nothing() - no value contained.

870 """

871

872 def __init__(self):

873 super().__init__(None)

Coverage for src/pyfplib/_core.py: 99%

222 statements