🐛 Fix sequential stream consuming source eager

[garlontas]

Summary by Sourcery

Fix the eager consumption issue in sequential streams by implementing lazy evaluation and enhance stream operations for better efficiency and correctness. Add comprehensive tests to validate stream behaviors, especially with generators.

Bug Fixes:

• Fix the eager consumption of source streams in sequential stream operations by ensuring lazy evaluation.

Enhancements:

• Refactor the concat method to execute all queued operations before concatenation, ensuring proper stream handling.
• Improve the distinct, drop_while, limit, skip, take_while, and any_match methods to handle streams more efficiently and correctly.

Tests:

• Add multiple tests to verify the correct behavior of stream concatenation, especially with generators and infinite streams.
• Enhance error handler tests to ensure proper iteration and exception handling.

[sourcery-ai]

Reviewer's Guide by Sourcery

This PR fixes issues with sequential stream consuming source eagerly by modifying the stream operations to work lazily with generators and infinite sequences. The changes primarily involve refactoring stream operations to use iterators instead of lists, and adding proper support for generator-based streams.

Updated class diagram for BaseStream

classDiagram
    class BaseStream {
        - _source: Iterable
        + concat(*streams: BaseStream[K]) BaseStream[K]
        + distinct() BaseStream[K]
        + drop_while(predicate: Callable[[K], bool]) BaseStream[K]
        + flat_map(mapper: Callable[[K], Iterable[_V]]) BaseStream[_V]
        + limit(max_size: int) BaseStream[K]
        + map(mapper: Callable[[K], _V]) BaseStream[_V]
        + skip(n: int) BaseStream[K]
        + take_while(predicate: Callable[[K], bool]) BaseStream[K]
        + any_match(predicate: Callable[[K], bool]) bool
        + count() int
        + min() Optional
        + max() Optional
    }
    note for BaseStream "Refactored to use iterators instead of lists for lazy evaluation"

Loading

Updated class diagram for SequentialStream

classDiagram
    class SequentialStream {
        + find_any() Optional
        + flat_map(mapper: Callable[[Any], BaseStream])
        + for_each(action: Callable)
        + map(mapper: Callable[[Any], Any])
        + peek(action: Callable)
        + reduce(predicate: Callable, identity, depends_on_state: bool)
    }
    note for SequentialStream "Updated flat_map to use flat_map function for lazy evaluation"

Loading

Updated class diagram for ParallelStream

classDiagram
    class ParallelStream {
        + find_any() Optional
        + flat_map(mapper: Callable[[Any], BaseStream])
    }
    note for ParallelStream "Updated flat_map to use mapper for lazy evaluation"

Loading

File-Level Changes

Change	Details	Files
Refactored stream operations to work lazily with iterators	Changed distinct() to use a generator-based implementation Modified limit() to work with iterators instead of slicing Updated skip() to use itertools.islice instead of list slicing Refactored take_while() to return iterator instead of materializing list Modified flat_map() to yield items instead of building list	pystreamapi/_itertools/tools.py pystreamapi/_streams/__base_stream.py
Fixed concat operation to properly handle streams	Changed concat to be an instance method instead of class method Fixed bug in concat that was dropping first stream's elements Added support for concatenating generator-based streams	pystreamapi/_streams/__base_stream.py pystreamapi/__stream.py
Added comprehensive tests for generator and infinite sequence handling	Added tests for concat with finite and infinite generators Added tests for distinct with infinite generators Added tests for skip, take_while, and any_match with infinite generators Added tests to verify proper handling of generator exhaustion	tests/_streams/test_base_stream.py tests/_streams/test_stream_implementation.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time. You can also use
  this command to specify where the summary should be inserted.

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

We've reviewed this pull request using the Sourcery rules engine. If you would also like our AI-powered code review then let us know.