async def process_source_chunks(
    source: BaseSource, weaviate_dest: WeaviateDestination, buffer_size: int = 50
):
    """Helper to fetch chunks from a given source connector and optionally insert them into Weaviate in bulk."""
    chunks_buffer = []
    async for chunk in source.generate_chunks():
        chunks_buffer.append(chunk)

        # When buffer is full, bulk insert
        if len(chunks_buffer) >= buffer_size:
            await weaviate_dest.bulk_insert(chunks_buffer)
            print(f"Inserted {len(chunks_buffer)} chunks from {source.__class__.__name__}")
            chunks_buffer.clear()

    # If any remain after the loop, insert them as well
    if chunks_buffer:
        await weaviate_dest.bulk_insert(chunks_buffer)
        print(f"Inserted final {len(chunks_buffer)} chunks from {source.__class__.__name__}")

    return chunks_buffer
Helper to fetch chunks from a given source connector and optionally insert them into Weaviate in bulk.
backend/local.py