Fix EngineClient.list_(jobs|programs) (#7848)

These methods need to use `_send_list_request_async` to page through
results in asyncio and collect them to a list before returning. This was
not caught by typechecks because these methods did not have type
annotations, so I've added some annotations.

In the future we could consider switching these to make the results
pageable, but would need to be careful about invoking the pager methods
in the correct event loop (see #7836).

Author: maffoo

cirq-google/cirq_google/engine/engine_client.py

@@ -211,7 +211,7 @@ async def list_programs_async(
         created_before: datetime.datetime | datetime.date | None = None,
         created_after: datetime.datetime | datetime.date | None = None,
         has_labels: dict[str, str] | None = None,
-    ):
+    ) -> list[quantum.QuantumProgram]:
         """Returns a list of previously executed quantum programs.
 
         Args:
@@ -242,7 +242,7 @@ async def list_programs_async(
         request = quantum.ListQuantumProgramsRequest(
             parent=_project_name(project_id), filter=" AND ".join(filters)
         )
-        return await self._send_request_async(self.grpc_client.list_quantum_programs, request)
+        return await self._send_list_request_async(self.grpc_client.list_quantum_programs, request)
 
     list_programs = duet.sync(list_programs_async)
 
@@ -485,7 +485,7 @@ async def list_jobs_async(
         execution_states: set[quantum.ExecutionStatus.State] | None = None,
         executed_processor_ids: list[str] | None = None,
         scheduled_processor_ids: list[str] | None = None,
-    ):
+    ) -> list[quantum.QuantumJob]:
         """Returns the list of jobs for a given program.
 
         Args:
@@ -545,7 +545,7 @@ async def list_jobs_async(
             program_id = "-"
         parent = _program_name_from_ids(project_id, program_id)
         request = quantum.ListQuantumJobsRequest(parent=parent, filter=" AND ".join(filters))
-        return await self._send_request_async(self.grpc_client.list_quantum_jobs, request)
+        return await self._send_list_request_async(self.grpc_client.list_quantum_jobs, request)
 
     list_jobs = duet.sync(list_jobs_async)
 

cirq-google/cirq_google/engine/engine_client_test.py

@@ -157,7 +157,7 @@ def test_list_program(client_constructor, default_engine_client):
         quantum.QuantumProgram(name='projects/proj/programs/prog1'),
         quantum.QuantumProgram(name='projects/proj/programs/prog2'),
     ]
-    grpc_client.list_quantum_programs.return_value = results
+    grpc_client.list_quantum_programs.return_value = _AsyncIterable(results)
 
     assert default_engine_client.list_programs(project_id='proj') == results
     grpc_client.list_quantum_programs.assert_called_with(
@@ -1252,7 +1252,7 @@ def test_list_jobs(client_constructor, default_engine_client):
         quantum.QuantumJob(name='projects/proj/programs/prog1/jobs/job1'),
         quantum.QuantumJob(name='projects/proj/programs/prog1/jobs/job2'),
     ]
-    grpc_client.list_quantum_jobs.return_value = results
+    grpc_client.list_quantum_jobs.return_value = _AsyncIterable(results)
 
     assert default_engine_client.list_jobs(project_id='proj', program_id='prog1') == results
     grpc_client.list_quantum_jobs.assert_called_with(
@@ -1265,6 +1265,15 @@ def test_list_jobs(client_constructor, default_engine_client):
     )
 
 
+class _AsyncIterable:
+    def __init__(self, items):
+        self.items = items
+
+    async def __aiter__(self):
+        for item in self.items:
+            yield item
+
+
 @pytest.mark.parametrize(
     'expected_filter, '
     'created_after, '