async def generate_entities(self) -> AsyncGenerator[ChunkEntity, None]:
        """Generate Elasticsearch index and document entities."""
        async with httpx.AsyncClient(base_url=self.url) as client:
            # Get all indices and their stats
            params = {
                "format": "json",
                "h": "index,health,status,docs.count,docs.deleted,store.size",
            }
            indices_data = await self._get(client, "/_cat/indices", params=params)

            # Determine which indices to sync
            allowed_indices = (
                [i.strip() for i in self.indices.split(",")]
                if self.indices and self.indices != "*"
                else None
            )

            for idx in indices_data:
                index_name = idx.get("index")
                if allowed_indices and index_name not in allowed_indices:
                    continue

                # Yield index metadata
                yield ElasticsearchIndexEntity(
                    entity_id=index_name,
                    index=index_name,
                    health=idx.get("health"),
                    status=idx.get("status"),
                    docs_count=int(idx.get("docs.count", 0)),
                    docs_deleted=int(idx.get("docs.deleted", 0)),
                    store_size=idx.get("store.size"),
                )

                # Fetch and yield documents using scroll API
                BATCH_SIZE = 100
                search_body = {"query": {"match_all": {}}, "size": BATCH_SIZE}
                if self.fields and self.fields != "*":
                    search_body["_source"] = [field.strip() for field in self.fields.split(",")]

                data = await self._post(client, f"/{index_name}/_search?scroll=1m", search_body)
                scroll_id = data["_scroll_id"]
                hits = data.get("hits", {}).get("hits", [])

                for hit in hits:
                    yield ElasticsearchDocumentEntity(
                        entity_id=f"{index_name}-{hit['_id']}",
                        index=index_name,
                        doc_id=hit["_id"],
                        source=hit.get("_source", {}),
                    )

                # Continue scrolling
                async for batch_hits in self._scroll_documents(client, scroll_id):
                    for hit in batch_hits:
                        yield ElasticsearchDocumentEntity(
                            entity_id=f"{index_name}-{hit['_id']}",
                            index=index_name,
                            doc_id=hit["_id"],
                            source=hit.get("_source", {}),
                        )
Generate Elasticsearch index and document entities.
backend/airweave/platform/sources/elasticsearch.py