Skip to content

omnipy.compute.helpers

CLASS DESCRIPTION
Void

Empty sync/async iterable for signature-only generator flow bodies.

Void

Empty sync/async iterable for signature-only generator flow bodies.

Use this when a flow template body exists only to define a generator or async-generator callable signature, while the child jobs perform the actual work.

Examples:

@LinearFlowTemplate(task_one, generator_task_two) def my_sync_generator_flow(number: int) -> Generator[int, None, None]: yield from Void() # For generator signature only; never run.

@DagFlowTemplate(task_one, generator_task_two) async def my_async_generator_flow(number: int) -> AsyncGenerator[int, None]: async for _ in Void(): # For generator signature only; never run. yield _

Source code in src/omnipy/compute/helpers.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class Void:
    """
    Empty sync/async iterable for signature-only generator flow bodies.

    Use this when a flow template body exists only to define a generator or
    async-generator callable signature, while the child jobs perform the
    actual work.

    Examples:
        @LinearFlowTemplate(task_one, generator_task_two)
        def my_sync_generator_flow(number: int) -> Generator[int, None, None]:
            yield from Void()  # For generator signature only; never run.

        @DagFlowTemplate(task_one, generator_task_two)
        async def my_async_generator_flow(number: int) -> AsyncGenerator[int, None]:
            async for _ in Void():  # For generator signature only; never run.
                yield _
    """
    def __iter__(self):
        return iter(())

    def __aiter__(self):
        return self

    async def __anext__(self):
        raise StopAsyncIteration