Coverage for typed_stream / _impl / stream.py: 99%

1# Licensed under the EUPL-1.2 or later. 

2# You may obtain a copy of the licence in all the official languages of the 

3# European Union at https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12 

4 

5"""Typed Stream class for easier handling of iterables.""" 

6 

7from __future__ import annotations 

8 

9import collections 

10import concurrent.futures 

11import functools 

12import itertools 

13import operator 

14import sys 

15import typing 

16from collections.abc import Callable, Iterable, Iterator, Mapping 

17from numbers import Number, Real 

18from types import EllipsisType 

19 

20from .. import exceptions, streamable 

21from . import _iteration_utils, functions, stream_abc 

22from ._default_value import DEFAULT_VALUE as _DEFAULT_VALUE 

23from ._default_value import DefaultValueType as _DefaultValueType 

24from ._typing import Self, TypeVarTuple, Unpack, override 

25from ._utils import map_with_additional_args 

26 

27# pylint: disable=too-many-lines 

28__all__: tuple[typing.Literal["Stream"]] = ("Stream",) 

29 

30if typing.TYPE_CHECKING: 

31 from . import _types, _utils 

32 

33K = typing.TypeVar("K") 

34T = typing.TypeVar("T") 

35U = typing.TypeVar("U") 

36V = typing.TypeVar("V") 

37Prim = typing.TypeVar("Prim", int, str, bool, complex, Number, Real) 

38Exc = typing.TypeVar("Exc", bound=BaseException) 

39 

40SA = typing.TypeVar("SA", bound="_types.SupportsAdd") 

41SC = typing.TypeVar("SC", bound="_types.SupportsComparison") 

42 

43Tvt = TypeVarTuple("Tvt") 

44 

45add: Callable[[SA, SA], SA] = operator.add 

46 

47 

48class Stream(stream_abc.StreamABC[T], Iterable[T]): 

49 """Typed Stream class for easier handling of iterables. 

50 

51 It is not recommended to store Stream instances in variables, 

52 instead use method chaining to handle the values and collect them when finished. 

53 """ 

54 

55 __slots__ = () 

56 

57 _data: Iterable[T] 

58 

59 def __init__( 

60 self, 

61 data: Iterable[T] | EllipsisType, 

62 close_source_callable: Callable[[], None] | None = None, 

63 ) -> None: 

64 """Create a new Stream. 

65 

66 To create a finished Stream do Stream(...). 

67 """ 

68 super().__init__( 

69 ... if isinstance(data, EllipsisType) else data, 

70 close_source_callable, 

71 ) 

72 

73 def __contains__(self, value: T, /) -> bool: 

74 """Check whether this stream contains the given value. 

75 

76 >>> 2 in Stream((1, 2, 3)) 

77 True 

78 >>> 4 in Stream((1, 2, 3)) 

79 False 

80 >>> 3 in Stream((1, 2, 3, 4, 5, 6, 7, 8)).peek(print) 

81 1 

82 2 

83 3 

84 True 

85 """ 

86 for element in self._data: 

87 if element == value: 

88 return self._finish(True, close_source=True) 

89 return self._finish(False, close_source=True) 

90 

91 @typing.overload 

92 def __getitem__(self, item: int, /) -> T: 

93 """Nobody inspects the spammish repetition.""" 

94 

95 @typing.overload 

96 def __getitem__( 

97 self, item: slice[int | None, int | None, int | None], / # noqa: W504 

98 ) -> streamable.StreamableSequence[T]: 

99 """Nobody inspects the spammish repetition.""" 

100 

101 def __getitem__( 

102 self, 

103 item: slice[int | None, int | None, int | None] | int, 

104 /, # noqa: W504 

105 ) -> streamable.StreamableSequence[T] | T: 

106 """Finish the stream by collecting. 

107 

108 >>> Stream((1, 2, 3))[1] 

109 2 

110 >>> Stream((1, 2, 3))[1:3] 

111 (2, 3) 

112 """ 

113 if isinstance(item, int): 

114 return self.nth(item) 

115 return self._get_slice(start=item.start, stop=item.stop, step=item.step) 

116 

117 @override 

118 def __iter__(self) -> Iterator[T]: 

119 """Iterate over the values of this Stream. This finishes the Stream. 

120 

121 >>> for value in Stream((1, 2, 3)): 

122 ... print(value) 

123 1 

124 2 

125 3 

126 """ 

127 return _iteration_utils.IterWithCleanUp(self._data, self.close) 

128 

129 def __length_hint__(self) -> int: 

130 """Return an estimated length for this Stream. 

131 

132 >>> from operator import length_hint 

133 >>> length_hint(Stream([1, 2, 3])) 

134 3 

135 >>> length_hint(Stream.range(100)) 

136 100 

137 """ 

138 return operator.length_hint(self._data) 

139 

140 def __reversed__(self) -> Iterator[T]: 

141 """Return the items of this Stream in reversed order. 

142 

143 This finishes the Stream and collects all the element. 

144 

145 Equivalent to reversed(self.collect()). 

146 

147 >>> tuple(reversed(Stream((1, 2, 3)))) 

148 (3, 2, 1) 

149 >>> "".join(reversed(Stream("abc"))) 

150 'cba' 

151 """ 

152 return reversed(self.collect()) 

153 

154 def _get_slice( # noqa: C901 

155 self, 

156 *, 

157 start: int | None = None, 

158 stop: int | None = None, 

159 step: int | None = None, 

160 ) -> streamable.StreamableSequence[T]: 

161 """Implement __getitem__ with slices.""" 

162 if start is stop is step is None: 

163 return self.collect() 

164 if ( # pylint: disable=too-many-boolean-expressions 

165 (start is None or start >= 0) 

166 and (step is None or step >= 0) 

167 and (stop is None or stop >= 0) 

168 ): 

169 return self._finish( 

170 streamable.StreamableSequence( 

171 itertools.islice(self._data, start, stop, step) 

172 ), 

173 close_source=True, 

174 ) 

175 if ( 

176 start is not None 

177 and start < 0 

178 and step in {None, 1} 

179 and stop is None 

180 ): 

181 return self.tail(abs(start)) 

182 return self.collect()[start:stop:step] 

183 

184 @classmethod 

185 @override 

186 def _module(cls) -> str: 

187 if cls == Stream: 

188 return "typed_stream" 

189 return cls.__module__ 

190 

191 @staticmethod 

192 def counting(start: int = 0, step: int = 1) -> Stream[int]: 

193 """Create an endless counting Stream. 

194 

195 >>> Stream.counting().limit(5).collect() 

196 (0, 1, 2, 3, 4) 

197 >>> Stream.counting(5, 2).limit(5).collect() 

198 (5, 7, 9, 11, 13) 

199 """ 

200 return Stream(itertools.count(start, step)) 

201 

202 @staticmethod 

203 def from_value(value: K) -> Stream[K]: 

204 """Create an endless Stream of the same value. 

205 

206 >>> Stream.from_value(1).limit(5).collect() 

207 (1, 1, 1, 1, 1) 

208 """ 

209 return Stream(itertools.repeat(value)) 

210 

211 @typing.overload 

212 @staticmethod 

213 def range(stop: int, /) -> Stream[int]: ... # noqa: D102 

214 

215 @typing.overload 

216 @staticmethod 

217 def range(*, stop: int) -> Stream[int]: ... # noqa: D102 

218 

219 @typing.overload 

220 @staticmethod 

221 def range(*, start: int, stop: int) -> Stream[int]: ... # noqa: D102 

222 

223 @typing.overload 

224 @staticmethod 

225 def range(start: int, stop: int, /) -> Stream[int]: ... # noqa: D102 

226 

227 @typing.overload 

228 @staticmethod 

229 def range(start: int, /, *, stop: int) -> Stream[int]: ... # noqa: D102 

230 

231 @typing.overload 

232 @staticmethod 

233 def range( # noqa: D102 

234 start: int, stop: int, /, *, step: int 

235 ) -> Stream[int]: ... 

236 

237 @typing.overload 

238 @staticmethod 

239 def range( # noqa: D102 

240 start: int, stop: int, step: int, / # noqa: W504 

241 ) -> Stream[int]: ... 

242 

243 @typing.overload 

244 @staticmethod 

245 def range( # noqa: D102 

246 start: int, /, *, stop: int, step: int 

247 ) -> Stream[int]: ... 

248 

249 @typing.overload 

250 @staticmethod 

251 def range( # noqa: D102 

252 *, start: int, stop: int, step: int 

253 ) -> Stream[int]: ... 

254 

255 @staticmethod 

256 def range( # noqa: C901 

257 *args: int, 

258 start: int | _DefaultValueType = _DEFAULT_VALUE, 

259 stop: int | _DefaultValueType = _DEFAULT_VALUE, 

260 step: int | _DefaultValueType = _DEFAULT_VALUE, 

261 ) -> Stream[int]: 

262 """Create a Stream[int] from a range. 

263 

264 The arguments behave like to the built-in range function: 

265 - Stream.range(stop) -> Stream[int] 

266 - Stream.range(start, stop[, step]) -> Stream[int] 

267 

268 >>> Stream.range(5).collect() == Stream(range(5)).collect() 

269 True 

270 >>> Stream.range(1, 13).collect() == Stream(range(1, 13)).collect() 

271 True 

272 >>> Stream.range(1, 9, 2).collect() == Stream(range(1, 9, 2)).collect() 

273 True 

274 >>> Stream.range(start=1, stop=7, step=2).collect() 

275 (1, 3, 5) 

276 """ 

277 # pylint: disable=confusing-consecutive-elif 

278 if not isinstance(start, _DefaultValueType): 

279 if not args and not isinstance(stop, _DefaultValueType): 

280 return Stream( 

281 range(start, stop) 

282 if isinstance(step, _DefaultValueType) 

283 else range(start, stop, step) 

284 ) 

285 elif isinstance(stop, _DefaultValueType): 

286 if isinstance(step, _DefaultValueType): 

287 return Stream(range(*args)) # no kwarg given 

288 if len(args) == 2: 

289 return Stream(range(args[0], args[1], step)) 

290 elif isinstance(step, _DefaultValueType): 

291 if len(args) == 1: 

292 return Stream(range(args[0], stop)) 

293 if not args: 

294 return Stream(range(stop)) 

295 elif len(args) == 1: 

296 return Stream(range(args[0], stop, step)) 

297 raise TypeError("Unexpected arguments to Stream.range()") 

298 

299 def all(self) -> bool: 

300 """Check whether all values are Truthy. This finishes the Stream. 

301 

302 Returns False if there is any false value in the Stream. 

303 

304 >>> Stream([1, 2, 3]).peek(print).all() 

305 1 

306 2 

307 3 

308 True 

309 >>> Stream([1, 2, 0, 4, 5, 6, 7, 8]).peek(print).all() 

310 1 

311 2 

312 0 

313 False 

314 >>> Stream([]).all() 

315 True 

316 """ 

317 return self._finish(all(self._data), close_source=True) 

318 

319 @typing.overload 

320 def catch( # noqa: D102 

321 self, 

322 *exception_class: type[Exc], 

323 ) -> Self: ... 

324 

325 @typing.overload 

326 def catch( # noqa: D102 

327 self, 

328 *exception_class: type[Exc], 

329 handler: Callable[[Exc], object], 

330 ) -> Self: ... 

331 

332 @typing.overload 

333 def catch( # noqa: D102 

334 self, 

335 *exception_class: type[Exc], 

336 default: Callable[[Exc], K] | Callable[[], K], 

337 ) -> Stream[T | K]: ... 

338 

339 @typing.overload 

340 def catch( # noqa: D102 

341 self, 

342 *exception_class: type[Exc], 

343 handler: Callable[[Exc], object], 

344 default: Callable[[Exc], K] | Callable[[], K], 

345 ) -> Stream[T | K]: ... 

346 

347 def catch( 

348 self, 

349 *exception_class: type[Exc], 

350 handler: Callable[[Exc], object] | None = None, 

351 default: Callable[[Exc], K] | Callable[[], K] | None = None, 

352 ) -> Stream[T | K]: 

353 """Catch exceptions. 

354 

355 >>> Stream("1a2").map(int).catch(ValueError, handler=print).collect() 

356 invalid literal for int() with base 10: 'a' 

357 (1, 2) 

358 >>> Stream("1a2").map(int).catch(ValueError, default=lambda _:_).collect() 

359 (1, ValueError("invalid literal for int() with base 10: 'a'"), 2) 

360 >>> Stream("1a2").map(int).peek(print) \ 

361 .catch(ValueError, handler=print).collect() 

362 1 

363 invalid literal for int() with base 10: 'a' 

364 2 

365 (1, 2) 

366 """ 

367 return self._finish( 

368 Stream( 

369 _iteration_utils.ExceptionHandler( 

370 self._data, exception_class, handler, default 

371 ) 

372 ) 

373 ) 

374 

375 def chain(self, *iterables: Iterable[T]) -> Self: 

376 """Add another iterable to the end of the Stream. 

377 

378 >>> Stream([1, 2, 3]).chain([4, 5, 6]).collect() 

379 (1, 2, 3, 4, 5, 6) 

380 >>> Stream([1, 2, 3]).chain([4, 5, 6], [7, 8, 9]).collect() 

381 (1, 2, 3, 4, 5, 6, 7, 8, 9) 

382 >>> Stream("abc").chain("def", "ghi", "jkl").collect("".join) 

383 'abcdefghijkl' 

384 """ 

385 self._data = itertools.chain(self._data, *iterables) 

386 return self 

387 

388 if sys.version_info >= (3, 12) and hasattr(itertools, "batched"): 

389 

390 def chunk(self, size: int) -> Stream[tuple[T, ...]]: 

391 """Split the stream into chunks of the specified size. 

392 

393 The last chunk may be shorter. 

394 

395 >>> Stream([1, 2, 3, 4, 5, 6]).chunk(2).collect() 

396 ((1, 2), (3, 4), (5, 6)) 

397 >>> Stream([1, 2, 3, 4, 5, 6]).chunk(3).collect() 

398 ((1, 2, 3), (4, 5, 6)) 

399 >>> Stream([1, 2, 3, 4, 5, 6, 7]).chunk(3).collect() 

400 ((1, 2, 3), (4, 5, 6), (7,)) 

401 """ 

402 return self._finish( 

403 Stream( 

404 # add strict=False if min version is 3.13 (is the default) 

405 itertools.batched(self._data, size), # noqa: B911 

406 ) 

407 ) 

408 

409 else: # pragma: no cover 

410 

411 def chunk(self, size: int) -> Stream[tuple[T, ...]]: 

412 """Split the stream into chunks of the specified size. 

413 

414 The last chunk may be shorter. 

415 

416 >>> Stream([1, 2, 3, 4, 5, 6]).chunk(2).collect() 

417 ((1, 2), (3, 4), (5, 6)) 

418 >>> Stream([1, 2, 3, 4, 5, 6]).chunk(3).collect() 

419 ((1, 2, 3), (4, 5, 6)) 

420 >>> Stream([1, 2, 3, 4, 5, 6, 7]).chunk(3).collect() 

421 ((1, 2, 3), (4, 5, 6), (7,)) 

422 """ 

423 return self._finish( 

424 _iteration_utils.Chunked(self._data, size).stream() 

425 ) 

426 

427 @typing.overload 

428 def collect(self, /) -> streamable.StreamableSequence[T]: ... # noqa: D102 

429 

430 @typing.overload 

431 def collect( # noqa: D102 

432 self, 

433 fun: Callable[[Iterable[T]], streamable.StreamableSequence[T]], 

434 /, 

435 ) -> streamable.StreamableSequence[T]: ... 

436 

437 @typing.overload 

438 def collect( # noqa: D102 

439 self, 

440 fun: type[collections.Counter[T]], 

441 /, 

442 ) -> collections.Counter[T]: ... 

443 

444 @typing.overload 

445 def collect( # noqa: D102 

446 self, fun: Callable[[Iterable[T]], tuple[T, ...]], / # noqa: W504 

447 ) -> tuple[T, ...]: ... 

448 

449 @typing.overload 

450 def collect( # noqa: D102 

451 self, fun: Callable[[Iterable[T]], list[T]], / # noqa: W504 

452 ) -> list[T]: ... 

453 

454 @typing.overload 

455 def collect( # noqa: D102 

456 self: Stream[SA], fun: Callable[[Iterable[SA]], SA], / # noqa: W504 

457 ) -> SA: ... 

458 

459 @typing.overload 

460 def collect( # noqa: D102 

461 self, fun: Callable[[Iterable[T]], set[T]], / # noqa: W504 

462 ) -> set[T]: ... 

463 

464 @typing.overload 

465 def collect( # noqa: D102 

466 self, fun: Callable[[Iterable[T]], frozenset[T]], / # noqa: W504 

467 ) -> frozenset[T]: ... 

468 

469 @typing.overload 

470 def collect( # noqa: D102 

471 self: Stream[tuple[K, V]], 

472 fun: Callable[[Iterable[tuple[K, V]]], dict[K, V]], 

473 /, 

474 ) -> dict[K, V]: ... 

475 

476 @typing.overload 

477 def collect( # noqa: D102 

478 self: Stream[tuple[K, V]], 

479 fun: Callable[[Iterable[tuple[K, V]]], Mapping[K, V]], 

480 /, 

481 ) -> Mapping[K, V]: ... 

482 

483 @typing.overload 

484 def collect( # noqa: D102 

485 self, fun: Callable[[Iterable[T]], K], / # noqa: W504 

486 ) -> K: ... 

487 

488 def collect( 

489 self: Stream[U], 

490 fun: Callable[[Iterable[U]], object] = streamable.StreamableSequence, 

491 /, 

492 ) -> object: 

493 """Collect the values of this Stream. This finishes the Stream. 

494 

495 >>> Stream([1, 2, 3]).collect(list) 

496 [1, 2, 3] 

497 >>> Stream([1, 2, 3]).collect(sum) 

498 6 

499 >>> Stream([1, 2, 3]).collect(dict.fromkeys) 

500 {1: None, 2: None, 3: None} 

501 >>> Stream([(1, 2), (3, 4)]).collect(dict) 

502 {1: 2, 3: 4} 

503 """ 

504 return self._finish(fun(self._data), close_source=True) 

505 

506 def concurrent_map( 

507 self, fun: Callable[[T], K], /, max_workers: int | None = None 

508 ) -> Stream[K]: 

509 """Map values concurrently. 

510 

511 See: https://docs.python.org/3/library/concurrent.futures.html 

512 

513 >>> Stream("123").concurrent_map(int).collect() 

514 (1, 2, 3) 

515 """ 

516 with concurrent.futures.ProcessPoolExecutor( 

517 max_workers=max_workers 

518 ) as executor: 

519 return self._finish(Stream(executor.map(fun, self._data))) 

520 

521 def conditional_map( 

522 self, 

523 condition: Callable[[T], object], 

524 if_true: Callable[[T], U], 

525 if_false: Callable[[T], V] | None = None, 

526 ) -> Stream[U | V]: 

527 """Map values conditionally. 

528 

529 >>> Stream("1x2x3x").conditional_map(str.isdigit, int).collect() 

530 (1, 2, 3) 

531 >>> Stream("1x2x3x").conditional_map(str.isdigit, int, ord).collect() 

532 (1, 120, 2, 120, 3, 120) 

533 """ 

534 return self._finish( 

535 Stream( 

536 _iteration_utils.IfElseMap( 

537 self._data, condition, if_true, if_false 

538 ) 

539 ) 

540 ) 

541 

542 def count(self) -> int: 

543 """Count the elements in this Stream. This finishes the Stream. 

544 

545 Equivalent to: Stream(...).map(lambda x: 1).sum() 

546 

547 >>> Stream([1, 2, 3]).count() 

548 3 

549 >>> Stream("abcdef").count() 

550 6 

551 """ 

552 return self._finish( 

553 _iteration_utils.count(self._data), close_source=True 

554 ) 

555 

556 def dedup(self, *, key: None | Callable[[T], object] = None) -> Self: 

557 """Remove consecutive equal values. 

558 

559 If the input is sorted this is the same as Stream.distinct(). 

560 

561 >>> Stream([1] * 100).dedup().collect(list) 

562 [1] 

563 >>> Stream([1, 2, 3, 1]).dedup().collect() 

564 (1, 2, 3, 1) 

565 >>> Stream([1, 1, 2, 2, 2, 2, 3, 1]).dedup().collect() 

566 (1, 2, 3, 1) 

567 >>> Stream([]).dedup().collect() 

568 () 

569 >>> Stream("ABC").dedup(key=str.lower).collect("".join) 

570 'ABC' 

571 >>> Stream("aAaAbbbcCCAaBbCc").dedup(key=str.lower).collect("".join) 

572 'abcABC' 

573 """ 

574 # Inspired by the unique_justseen itertools recipe 

575 # https://docs.python.org/3/library/itertools.html#itertools-recipes 

576 self._data = map( 

577 next, 

578 map( 

579 operator.itemgetter(1), 

580 itertools.groupby(self._data, key=key), 

581 ), 

582 ) 

583 return self 

584 

585 def dedup_counting(self) -> Stream[tuple[T, int]]: 

586 """Group the stream and count the items in the group. 

587 

588 >>> Stream("abba").dedup_counting().starmap(print).for_each() 

589 a 1 

590 b 2 

591 a 1 

592 >>> Stream("AaaaBBcccc").dedup_counting().starmap(print).for_each() 

593 A 1 

594 a 3 

595 B 2 

596 c 4 

597 """ 

598 

599 def _map(k: T, g: Iterator[T]) -> tuple[T, int]: 

600 return (k, _iteration_utils.count(g)) 

601 

602 return Stream(itertools.starmap(_map, itertools.groupby(self))) 

603 

604 @override 

605 def distinct(self, *, use_set: bool = True) -> Self: 

606 """Remove duplicate values. 

607 

608 >>> from typed_stream import Stream 

609 >>> Stream([1, 2, 2, 2, 3, 2, 2]).distinct().collect() 

610 (1, 2, 3) 

611 >>> Stream([{1}, {2}, {3}, {2}, {2}]).distinct().collect() # doctest: +ELLIPSIS 

612 Traceback (most recent call last): 

613 ... 

614 TypeError: ...unhashable type: 'set'... 

615 >>> Stream([{1}, {2}, {3}, {2}, {2}]).distinct(use_set=False).collect() 

616 ({1}, {2}, {3}) 

617 """ 

618 # pylint: disable=duplicate-code 

619 encountered: set[T] | list[T] 

620 peek_fun: Callable[[T], None] 

621 if use_set: 

622 encountered = set() 

623 peek_fun = encountered.add 

624 else: 

625 encountered = [] 

626 peek_fun = encountered.append 

627 

628 self._data = map( 

629 _iteration_utils.Peeker(peek_fun), 

630 itertools.filterfalse(encountered.__contains__, self._data), 

631 ) 

632 return self 

633 

634 @override 

635 def drop(self, c: int, /) -> Self: 

636 """Drop the first count values. 

637 

638 >>> Stream([1, 2, 3, 4, 5]).drop(2).collect() 

639 (3, 4, 5) 

640 """ 

641 self._data = itertools.islice(self._data, c, None) 

642 return self 

643 

644 def drop_while(self, fun: Callable[[T], object], /) -> Self: 

645 """Drop values as long as the function returns a truthy value. 

646 

647 See: https://docs.python.org/3/library/itertools.html#itertools.dropwhile 

648 

649 >>> Stream([1, 2, 3, 4, 1]).drop_while(lambda x: x < 3).collect() 

650 (3, 4, 1) 

651 """ 

652 self._data = itertools.dropwhile(fun, self._data) 

653 return self 

654 

655 def empty(self) -> bool: 

656 """Check whether this doesn't contain any value. This finishes the Stream. 

657 

658 >>> Stream([1, 2, 3]).empty() 

659 False 

660 >>> Stream([]).empty() 

661 True 

662 """ 

663 try: 

664 self.first() 

665 except exceptions.StreamEmptyError: 

666 return True 

667 return False 

668 

669 def enumerate( 

670 self, start_index: int = 0, / # noqa: W504 

671 ) -> Stream[_utils.IndexValueTuple[T]]: 

672 """Map the values to a tuple of index and value. 

673 

674 >>> Stream([1, 2, 3]).enumerate().collect() 

675 ((0, 1), (1, 2), (2, 3)) 

676 >>> Stream("abc").enumerate().collect() 

677 ((0, 'a'), (1, 'b'), (2, 'c')) 

678 >>> Stream("abc").enumerate(100).collect() 

679 ((100, 'a'), (101, 'b'), (102, 'c')) 

680 >>> Stream("abc").enumerate().map(lambda el: {el.idx: el.val}).collect() 

681 ({0: 'a'}, {1: 'b'}, {2: 'c'}) 

682 """ 

683 return self._finish( 

684 Stream(_iteration_utils.Enumerator(self._data, start_index)) 

685 ) 

686 

687 @typing.overload 

688 def exclude( # noqa: D102 

689 self: Stream[K | Prim], 

690 fun: _utils.InstanceChecker[Prim], 

691 /, # noqa: W504 

692 ) -> Stream[K]: ... 

693 

694 @typing.overload 

695 def exclude( # noqa: D102 

696 self: Stream[K | U], fun: _utils.InstanceChecker[U], / # noqa: W504 

697 ) -> Stream[K]: ... 

698 

699 @typing.overload 

700 def exclude( # noqa: D102 

701 self: Stream[K | None], fun: _utils.NoneChecker, / # noqa: W504 

702 ) -> Stream[K]: ... 

703 

704 # @typing.overload 

705 # def exclude( # noqa: D102 

706 # self: Stream[K | U], fun: TypeGuardingCallable[U, K | U] 

707 # ) -> Stream[K]: 

708 # ... 

709 

710 @typing.overload 

711 def exclude( # noqa: D102 

712 self: Stream[T], fun: Callable[[T], object], / # noqa: W504 

713 ) -> Stream[T]: ... 

714 

715 @override 

716 def exclude(self, fun: Callable[[T], object], /) -> object: 

717 """Exclude values if the function returns a truthy value. 

718 

719 See: https://docs.python.org/3/library/itertools.html#itertools.filterfalse 

720 

721 >>> Stream([1, 2, 3, 4, 5]).exclude(lambda x: x % 2).collect() 

722 (2, 4) 

723 >>> Stream([1, 2, None, 3]).exclude(lambda x: x is None).collect() 

724 (1, 2, 3) 

725 """ 

726 self._data = itertools.filterfalse(fun, self._data) 

727 return self 

728 

729 @typing.overload 

730 def filter( # noqa: D102 

731 self: Stream[K | None], fun: _utils.NotNoneChecker 

732 ) -> Stream[K]: ... 

733 

734 @typing.overload 

735 def filter(self: Stream[K | None]) -> Stream[K]: ... # noqa: D102 

736 

737 @typing.overload 

738 def filter(self: Stream[T]) -> Stream[T]: ... # noqa: D102 

739 

740 @typing.overload 

741 def filter( # noqa: D102 

742 self, fun: _utils.InstanceChecker[K] 

743 ) -> Stream[K]: # pragma: no cover 

744 ... 

745 

746 @typing.overload 

747 def filter( # noqa: D102 

748 self, fun: _types.TypeGuardingCallable[K, T] 

749 ) -> Stream[K]: # pragma: no cover 

750 ... 

751 

752 @typing.overload 

753 def filter( # noqa: D102 

754 self: Stream[T], fun: Callable[[T], object] 

755 ) -> Stream[T]: ... 

756 

757 def filter(self, fun: Callable[[T], object] | None = None) -> object: 

758 """Use built-in filter to filter values. 

759 

760 >>> Stream([1, 2, 3, 4, 5]).filter(lambda x: x % 2).collect() 

761 (1, 3, 5) 

762 """ 

763 self._data = filter(fun, self._data) 

764 return self 

765 

766 def first(self, default: K | _DefaultValueType = _DEFAULT_VALUE) -> T | K: 

767 """Return the first element of the Stream. This finishes the Stream. 

768 

769 >>> Stream([1, 2, 3]).first() 

770 1 

771 >>> Stream("abc").first() 

772 'a' 

773 >>> Stream([]).first() 

774 Traceback (most recent call last): 

775 ... 

776 typed_stream.exceptions.StreamEmptyError 

777 >>> Stream([]).first(default="default") 

778 'default' 

779 """ 

780 try: 

781 first = next(iter(self._data)) 

782 except StopIteration: 

783 if not isinstance(default, _DefaultValueType): 

784 return default 

785 raise exceptions.StreamEmptyError() from None 

786 finally: 

787 self._finish(None, close_source=True) 

788 return first 

789 

790 @typing.overload 

791 def flat_map( # noqa: D102 

792 self, fun: Callable[[T], Iterable[K]], / # noqa: W504 

793 ) -> Stream[K]: ... 

794 

795 @typing.overload 

796 def flat_map( # noqa: D102 

797 self, 

798 fun: Callable[[T, Unpack[Tvt]], Iterable[K]], 

799 /, 

800 *args: Unpack[Tvt], 

801 ) -> Stream[K]: ... 

802 

803 def flat_map( 

804 self, 

805 fun: Callable[[T, Unpack[Tvt]], Iterable[K]], 

806 /, 

807 *args: Unpack[Tvt], 

808 ) -> Stream[K]: 

809 """Map each value to another. 

810 

811 This lazily finishes the current Stream and creates a new one. 

812 

813 >>> Stream([1, 4, 7]).flat_map(lambda x: [x, x + 1, x + 2]).collect() 

814 (1, 2, 3, 4, 5, 6, 7, 8, 9) 

815 >>> Stream(["abc", "def"]).flat_map(str.encode, "ASCII").collect() 

816 (97, 98, 99, 100, 101, 102) 

817 """ 

818 return Stream( 

819 itertools.chain.from_iterable( 

820 map_with_additional_args(self._data, fun, args) 

821 ) 

822 ) 

823 

824 def for_each(self, fun: Callable[[T], object] = functions.noop, /) -> None: 

825 """Consume all the values of the Stream with the callable. 

826 

827 >>> Stream([1, 2, 3]).for_each(print) 

828 1 

829 2 

830 3 

831 """ 

832 for value in self._data: 

833 fun(value) 

834 self._finish(None, close_source=True) 

835 

836 def last(self) -> T: 

837 """Return the last element of the Stream. This finishes the Stream. 

838 

839 raises StreamEmptyError if stream is empty. 

840 

841 >>> Stream([1, 2, 3]).last() 

842 3 

843 >>> Stream([]).last() 

844 Traceback (most recent call last): 

845 ... 

846 typed_stream.exceptions.StreamEmptyError 

847 """ 

848 if tail := self.tail(1): 

849 return tail[-1] 

850 raise exceptions.StreamEmptyError() 

851 

852 @override 

853 def limit(self, c: int, /) -> Self: 

854 """Stop the Stream after count values. 

855 

856 >>> Stream([1, 2, 3, 4, 5]).limit(3).collect() 

857 (1, 2, 3) 

858 >>> Stream.from_value(3).limit(1000).collect() == (3,) * 1000 

859 True 

860 """ 

861 self._data = itertools.islice(self._data, c) 

862 return self 

863 

864 @typing.overload 

865 def map(self, fun: Callable[[T], K], /) -> Stream[K]: ... # noqa: D102