aiostream 0.2.0

Generator-based operators for asynchronous iteration

Generator-based operators for asynchronous iteration

Synopsis

This library provides a collection of stream operators that can be combined to create asynchronous pipelines of operations.

It can be seen as an asynchronous version of itertools, although some aspects are slightly different. Essentially, all the provided operators return a unified interface called a stream. A stream is an enhanced asynchronous iterable providing the following features:

• Pipe-lining of operators - using pipe symbol |

• Repeatability - every iteration creates a different iterator

• Safe iteration context - using async with statement and the stream method

• Simplified execution - get the last element from a steam using await statement

• Slicing and indexing - using square brackets []

• Concatenation - using addition symbol +

Requirements

The stream operators rely heavily on asynchronous generators (PEP 525):

• python >= 3.6

Example

The following example demonstrates most of the streams capabilities:

import asyncio
from aiostream import stream, pipe


async def main():

    # Create a counting stream with a 0.2 seconds interval
    xs = stream.count(interval=0.2)

    # Operators can be piped using '|'
    ys = xs | pipe.map(lambda x: x**2)

    # Streams can be sliced
    zs = ys[1:10:2]

    # Use a stream context for proper resource management
    async with zs.stream() as streamer:

        # Asynchronous iteration
        async for z in streamer:

            # Print 1, 9, 25, 49 and 81
            print('->', z)

    # Streams can be awaited and return the last value
    print('9² = ', await zs)

    # Streams can run several times
    print('9² = ', await zs)

    # Streams can be concatenated
    one_two_three = stream.just(1) + stream.range(2, 4)

    # Print [1, 2, 3]
    print(await stream.list(one_two_three))


# Run main coroutine
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Operators

Creation operators (non-pipable)

• iterate (it):

  Generate values from a synchronous or asynchronous iterable.

• preserve (ait):

  Generate values from an asynchronous iterable without explicitly closing the corresponding iterator.

• just (value):

  Generate a single value.

• empty ():

  Terminate without generating any value.

• throw (exc):

  Throw an exception without generating any value.

• never ():

  Hang forever without generating any value.

• repeat (value, times=None, *, interval=0):

  Generate the same value a given number of times.

• range (*args, interval=0):

  Generate a given range of numbers.

• count (start=0, step=1, *, interval=0):

  Generate consecutive numbers indefinitely.

Transformation operators

• map (source, func, *more_sources):

  Apply a given function to the elements of one or several asynchronous sequences.

• enumerate (source, start=0, step=1):

  Generate (index, value) tuples from an asynchronous sequence.

• starmap (source, func):

  Apply a given function to the unpacked elements of an asynchronous sequence.

• cycle (source):

  Iterate indefinitely over an asynchronous sequence.

Selection operators

• take (source, n):

  Forward the first n elements from an asynchronous sequence.

• take_last (source, n):

  Forward the last n elements from an asynchronous sequence.

• skip (source, n):

  Forward an asynchronous sequence, skipping the first n elements.

• skip_last (source, n):

  Forward an asynchronous sequence, skipping the last n elements.

• filter_index (source, func):

  Filter an asynchronous sequence using the index of the elements.

• slice (source, *args):

  Slice an asynchronous sequence.

• item_at (source, index):

  Forward the nth element of an asynchronous sequence.

• get_item (source, index):

  Forward one or several items from an asynchronous sequence.

• filter (source, func):

  Filter an asynchronous sequence using an arbitrary function.

• takewhile (source, func):

  Forward an asynchronous sequence while a condition is met.

• dropwhile (source, func):

  Discard the elements from an asynchronous sequence while a condition is met.

Combination operators

• map (source, func, *more_sources):

  Apply a given function to the elements of one or several asynchronous sequences.

• zip (*sources):

  Combine and forward the elements of several asynchronous sequences.

• merge (*sources):

  Merge several asynchronous sequences together.

• chain (*sources):

  Chain asynchronous sequences together, in the order they are given.

Aggregatation operators

• accumulate (source, func=op.add, initializer=None):

  Generate a series of accumulated sums (or other binary function) from an asynchronous sequence.

• reduce (source, func, initializer=None):

  Apply a function of two arguments cumulatively to the items of an asynchronous sequence, reducing the sequence to a single value.

• list ():

  Generate a single list from an asynchronous sequence.

Timing operators

• space_out (source, interval):

  Make sure the elements of an asynchronous sequence are separated in time by the given interval.

• timeout (source, timeout):

  Raise a time-out if an element of the asynchronous sequence takes too long to arrive.

• delay (source, delay):

  Delay the iteration of an asynchronous sequence.

Misc operators

• action (source, func):

  Perform an action for each element of an asynchronous sequence without modifying it.

• print (source, template=None, **kwargs):

  Print each element of an asynchronous sequence without modifying it.

References

This library is inspired by:

• PEP 525: Asynchronous Generators

• Rx - Reactive Extensions

• aioreactive - Async/await reactive tools for Python 3.5+

• itertools - Functions creating iterators for efficient looping

Contact

Vincent Michel: [e-mail]