Data Source Aggregator

Build a data source aggregator that can uniformly process data from various sources regardless of whether they are synchronous or asynchronous iterables. The system should accept different types of iterables and process them in a consistent manner.

Requirements

Input Processing

The system should accept data from multiple source types:

• Regular Python lists and tuples (synchronous iterables)
• Async generator functions
• Async iterator objects
• Any combination of the above

Core Functionality

Implement a function aggregate_data(sources) that:

• Accepts a list of various iterable sources (both sync and async)
• Processes each source uniformly regardless of its type
• Collects all items from all sources into a single list
• Returns the aggregated results

Data Transformation

Implement a function transform_sources(sources, transformer) that:

• Accepts a list of various iterable sources
• Applies a transformer function to each item from all sources
• The transformer can be either synchronous or asynchronous
• Returns a list of transformed items

Error Handling

• Handle empty sources gracefully (return empty list)
• If a source is None, skip it and continue processing other sources

API

async def aggregate_data(sources):
    """
    Aggregate data from multiple sources of varying types.

    Args:
        sources: List of iterable sources (sync or async)

    Returns:
        List containing all items from all sources
    """
    pass

async def transform_sources(sources, transformer):
    """
    Apply a transformer to items from multiple sources.

    Args:
        sources: List of iterable sources (sync or async)
        transformer: Function to apply to each item (sync or async)

    Returns:
        List of transformed items
    """
    pass

Implementation

@generates

Tests

• Given a list containing both a regular list and an async generator, aggregate_data() returns all items from both sources @test
• Given multiple sources including None, aggregate_data() skips None sources and processes valid ones @test
• Given empty sources, aggregate_data() returns an empty list @test
• Given sources with a synchronous transformer function, transform_sources() applies the transformation to all items @test

Dependencies { .dependencies }

asyncstdlib { .dependency }

Provides async-compatible utilities for working with async iterables uniformly.