async def scan_aggs(
    search: AsyncSearch,
    source_aggs: Sequence[Mapping[str, Agg]],
    inner_aggs: Dict[str, Agg] = {},
    size: int = 10,
) -> AsyncIterator[CompositeAggregate]:
    """
    Helper function used to iterate over all possible bucket combinations of
    ``source_aggs``, returning results of ``inner_aggs`` for each. Uses the
    ``composite`` aggregation under the hood to perform this.
    """

    async def run_search(**kwargs: Any) -> Response:
        s = search[:0]
        bucket = s.aggs.bucket(
            "comp",
            aggs.Composite(
                sources=source_aggs,
                size=size,
                **kwargs,
            ),
        )
        for agg_name, agg in inner_aggs.items():
            bucket[agg_name] = agg
        return await s.execute()

    response = await run_search()
    while response.aggregations["comp"].buckets:
        for b in response.aggregations["comp"].buckets:
            yield cast(CompositeAggregate, b)
        if "after_key" in response.aggregations["comp"]:
            after = response.aggregations["comp"].after_key
        else:
            after = response.aggregations["comp"].buckets[-1].key
        response = await run_search(after=after)

    Helper function used to iterate over all possible bucket combinations of
    ``source_aggs``, returning results of ``inner_aggs`` for each. Uses the
    ``composite`` aggregation under the hood to perform this.
    
examples/dsl/async/composite_agg.py