feat: add batch nested resource access across all resources

Enable .files, .runs, .documents, and .datasets on batch resources,
supporting concurrent .list() fan-out across multiple parent IDs.

Author: alphamt

README.md

@@ -84,6 +84,10 @@ async def main():
     # Batch operations
     results = await client.v1.stages("id-1", "id-2", "id-3").get(concurrency=5)
 
+    # Batch nested resources — fan out .list() across multiple parents
+    all_files = await client.v1.stages("id-1", "id-2").files.list()
+    all_runs  = await client.v1.stages("id-1", "id-2").runs.list()
+
 asyncio.run(main())
 ```
 
@@ -116,6 +120,12 @@ await client.v1.stages("stage-id").delete()
 stages = await client.v1.stages("id-1", "id-2").get()
 await client.v1.stages("id-1", "id-2").delete()
 
+# Batch nested resources — fan out .list() concurrently
+all_files = await client.v1.stages("id-1", "id-2").files.list()
+all_runs  = await client.v1.stages("id-1", "id-2").runs.list()
+all_docs  = await client.v1.search_stores("ss-1", "ss-2").documents.list()
+all_ds    = await client.v1.repositories("r-1", "r-2").datasets.list()
+
 # Nested resources
 file_content = await client.v1.stages("stage-id").files("file-id").get()
 presigned    = await client.v1.stages("stage-id").files("file-id").presigned_url(ttl=3600)

pharia/resources/connectors.py

@@ -46,6 +46,44 @@ async def list(self, page: int = 0, size: int = 100, status: str = "") -> RunLis
         )
 
 
+@dataclass
+class BatchConnectorFiles:
+    """Batch list operations for files across multiple connectors."""
+
+    client: "Client"
+    connector_ids: list[str]
+
+    async def list(
+        self, page: int = 0, size: int = 100, concurrency: int = 10
+    ) -> list[ConnectorFilesListResponse]:
+        """List files in multiple connectors concurrently."""
+        coros = [
+            ConnectorFiles(client=self.client, connector_id=cid).list(page=page, size=size)
+            for cid in self.connector_ids
+        ]
+        return await gather_with_limit(coros, concurrency)
+
+
+@dataclass
+class BatchConnectorRuns:
+    """Batch list operations for runs across multiple connectors."""
+
+    client: "Client"
+    connector_ids: list[str]
+
+    async def list(
+        self, page: int = 0, size: int = 100, status: str = "", concurrency: int = 10
+    ) -> list[RunListResponse]:
+        """List runs in multiple connectors concurrently."""
+        coros = [
+            ConnectorRuns(client=self.client, connector_id=cid).list(
+                page=page, size=size, status=status
+            )
+            for cid in self.connector_ids
+        ]
+        return await gather_with_limit(coros, concurrency)
+
+
 @dataclass
 class ConnectorResource:
     """Instance-level operations for a single connector."""
@@ -79,6 +117,16 @@ class BatchConnectorResource:
     client: "Client"
     connector_ids: list[str]
 
+    @property
+    def files(self) -> BatchConnectorFiles:
+        """Access files across multiple connectors."""
+        return BatchConnectorFiles(client=self.client, connector_ids=self.connector_ids)
+
+    @property
+    def runs(self) -> BatchConnectorRuns:
+        """Access runs across multiple connectors."""
+        return BatchConnectorRuns(client=self.client, connector_ids=self.connector_ids)
+
     async def get(self, concurrency: int = 10) -> list[Connector]:
         """Retrieve multiple connectors concurrently."""
         coros = [self.client.request("GET", f"/connectors/{cid}") for cid in self.connector_ids]

pharia/resources/datasets.py

@@ -143,3 +143,33 @@ async def create(self, **dataset_data: Unpack[CreateDatasetInput]) -> Dataset:
         return await self.client.request(
             "POST", f"/repositories/{self.repository_id}/datasets", json=payload
         )
+
+
+@dataclass
+class BatchRepositoryDatasets:
+    """Batch list operations for datasets across multiple repositories."""
+
+    client: "Client"
+    repository_ids: list[str]
+
+    async def list(
+        self,
+        page: int = 0,
+        size: int = 100,
+        label: list[str] | None = None,
+        created_after: str = "",
+        created_before: str = "",
+        concurrency: int = 10,
+    ) -> list[DatasetListResponse]:
+        """List datasets in multiple repositories concurrently."""
+        coros = [
+            RepositoryDatasets(client=self.client, repository_id=rid).list(
+                page=page,
+                size=size,
+                label=label,
+                created_after=created_after,
+                created_before=created_before,
+            )
+            for rid in self.repository_ids
+        ]
+        return await gather_with_limit(coros, concurrency)

pharia/resources/documents.py

@@ -122,3 +122,28 @@ async def list(
         return await self.client.request(
             "GET", f"/search_stores/{self.search_store_id}/documents", params=params
         )
+
+
+@dataclass
+class BatchSearchStoreDocuments:
+    """Batch list operations for documents across multiple search stores."""
+
+    client: "Client"
+    search_store_ids: list[str]
+
+    async def list(
+        self,
+        page: int = 1,
+        size: int = 100,
+        name: str = "",
+        starts_with: str = "",
+        concurrency: int = 10,
+    ) -> list[DocumentListResponse]:
+        """List documents in multiple search stores concurrently."""
+        coros = [
+            SearchStoreDocuments(client=self.client, search_store_id=sid).list(
+                page=page, size=size, name=name, starts_with=starts_with
+            )
+            for sid in self.search_store_ids
+        ]
+        return await gather_with_limit(coros, concurrency)

pharia/resources/files.py

@@ -109,3 +109,33 @@ async def list(
     async def create(self, file_data: dict) -> File:
         """Upload a file to this stage."""
         return await self.client.request("POST", f"/stages/{self.stage_id}/files", json=file_data)
+
+
+@dataclass
+class BatchStageFiles:
+    """Batch list operations for files across multiple stages."""
+
+    client: "Client"
+    stage_ids: list[str]
+
+    async def list(
+        self,
+        page: int = 0,
+        size: int = 100,
+        name: str = "",
+        created_after: str = "",
+        created_before: str = "",
+        concurrency: int = 10,
+    ) -> list[FileListResponse]:
+        """List files in multiple stages concurrently."""
+        coros = [
+            StageFiles(client=self.client, stage_id=sid).list(
+                page=page,
+                size=size,
+                name=name,
+                created_after=created_after,
+                created_before=created_before,
+            )
+            for sid in self.stage_ids
+        ]
+        return await gather_with_limit(coros, concurrency)

pharia/resources/repositories.py

@@ -8,7 +8,7 @@
 from pharia.models import RepositoryListResponse
 from pharia.models import create_repository_to_api
 from pharia.resources.base import gather_with_limit
-from pharia.resources.datasets import DatasetResource
+from pharia.resources.datasets import BatchRepositoryDatasets
 from pharia.resources.datasets import RepositoryDatasets
 
 
@@ -44,6 +44,11 @@ class BatchRepositoryResource:
     client: "Client"
     repository_ids: list[str]
 
+    @property
+    def datasets(self) -> BatchRepositoryDatasets:
+        """Access datasets across multiple repositories."""
+        return BatchRepositoryDatasets(client=self.client, repository_ids=self.repository_ids)
+
     async def get(self, concurrency: int = 10) -> list[Repository]:
         """Retrieve multiple repositories concurrently."""
         coros = [self.client.request("GET", f"/repositories/{rid}") for rid in self.repository_ids]

pharia/resources/search_stores.py

@@ -15,6 +15,7 @@
 from pharia.models import search_input_to_api
 from pharia.models import update_search_store_to_api
 from pharia.resources.base import gather_with_limit
+from pharia.resources.documents import BatchSearchStoreDocuments
 from pharia.resources.documents import SearchStoreDocuments
 
 
@@ -79,6 +80,11 @@ class BatchSearchStoreResource:
     client: "Client"
     search_store_ids: list[str]
 
+    @property
+    def documents(self) -> BatchSearchStoreDocuments:
+        """Access documents across multiple search stores."""
+        return BatchSearchStoreDocuments(client=self.client, search_store_ids=self.search_store_ids)
+
     async def get(self, concurrency: int = 10) -> list[SearchStore]:
         """Retrieve multiple search stores concurrently."""
         coros = [

pharia/resources/stages.py

@@ -18,6 +18,7 @@
 from pharia.models import create_stage_to_api
 from pharia.models import update_stage_to_api
 from pharia.resources.base import gather_with_limit
+from pharia.resources.files import BatchStageFiles
 from pharia.resources.files import StageFiles
 
 
@@ -38,6 +39,24 @@ async def list(self, page: int = 0, size: int = 100, status: str = "") -> RunLis
         return await self.client.request("GET", f"/stages/{self.stage_id}/runs", params=params)
 
 
+@dataclass
+class BatchStageRuns:
+    """Batch list operations for runs across multiple stages."""
+
+    client: "Client"
+    stage_ids: list[str]
+
+    async def list(
+        self, page: int = 0, size: int = 100, status: str = "", concurrency: int = 10
+    ) -> list[RunListResponse]:
+        """List runs in multiple stages concurrently."""
+        coros = [
+            StageRuns(client=self.client, stage_id=sid).list(page=page, size=size, status=status)
+            for sid in self.stage_ids
+        ]
+        return await gather_with_limit(coros, concurrency)
+
+
 @dataclass
 class StageResource:
     """Instance-level operations for a single stage."""
@@ -76,6 +95,16 @@ class BatchStageResource:
     client: "Client"
     stage_ids: list[str]
 
+    @property
+    def files(self) -> BatchStageFiles:
+        """Access files across multiple stages."""
+        return BatchStageFiles(client=self.client, stage_ids=self.stage_ids)
+
+    @property
+    def runs(self) -> BatchStageRuns:
+        """Access runs across multiple stages."""
+        return BatchStageRuns(client=self.client, stage_ids=self.stage_ids)
+
     async def get(self, concurrency: int = 10) -> list[Stage]:
         """Retrieve multiple stages concurrently."""
         coros = [self.client.request("GET", f"/stages/{sid}") for sid in self.stage_ids]

pharia/tests/test_stages.py

@@ -50,3 +50,65 @@ async def test_stages_nested_files():
     assert hasattr(file_resource, "update")
     assert hasattr(file_resource, "delete")
     assert hasattr(file_resource, "presigned_url")
+
+
+@pytest.mark.asyncio
+async def test_batch_stages_has_files_and_runs():
+    """Test batch stages expose .files and .runs with .list() methods."""
+    client = Client(base_url="https://api.example.com", api_key="test-key")
+
+    batch = client.v1.stages("id1", "id2")
+    assert hasattr(batch, "files")
+    assert hasattr(batch, "runs")
+
+    batch_files = batch.files
+    assert hasattr(batch_files, "list")
+    assert batch_files.stage_ids == ["id1", "id2"]
+
+    batch_runs = batch.runs
+    assert hasattr(batch_runs, "list")
+    assert batch_runs.stage_ids == ["id1", "id2"]
+
+
+@pytest.mark.asyncio
+async def test_batch_search_stores_has_documents():
+    """Test batch search stores expose .documents with .list() method."""
+    client = Client(base_url="https://api.example.com", api_key="test-key")
+
+    batch = client.v1.search_stores("ss1", "ss2")
+    assert hasattr(batch, "documents")
+
+    batch_docs = batch.documents
+    assert hasattr(batch_docs, "list")
+    assert batch_docs.search_store_ids == ["ss1", "ss2"]
+
+
+@pytest.mark.asyncio
+async def test_batch_repositories_has_datasets():
+    """Test batch repositories expose .datasets with .list() method."""
+    client = Client(base_url="https://api.example.com", api_key="test-key")
+
+    batch = client.v1.repositories("r1", "r2")
+    assert hasattr(batch, "datasets")
+
+    batch_datasets = batch.datasets
+    assert hasattr(batch_datasets, "list")
+    assert batch_datasets.repository_ids == ["r1", "r2"]
+
+
+@pytest.mark.asyncio
+async def test_batch_connectors_has_files_and_runs():
+    """Test batch connectors expose .files and .runs with .list() methods."""
+    client = Client(base_url="https://api.example.com", api_key="test-key")
+
+    batch = client.v1.connectors("c1", "c2")
+    assert hasattr(batch, "files")
+    assert hasattr(batch, "runs")
+
+    batch_files = batch.files
+    assert hasattr(batch_files, "list")
+    assert batch_files.connector_ids == ["c1", "c2"]
+
+    batch_runs = batch.runs
+    assert hasattr(batch_runs, "list")
+    assert batch_runs.connector_ids == ["c1", "c2"]

pyproject.toml

@@ -167,3 +167,6 @@ addopts = [
     "--strict-config",
     "--showlocals",
 ]
+
+[dependency-groups]
+dev = []