code: def ReadFlowRequestsReadyForProcessing(self,
                                           client_id,
                                           flow_id,
                                           next_needed_request,
                                           cursor=None):
      
      query = ("SELECT request, needs_processing, responses_expected, "
               "UNIX_TIMESTAMP(timestamp) "
               "FROM flow_requests "
               "WHERE client_id=%s AND flow_id=%s")
      args = [db_utils.ClientIDToInt(client_id), db_utils.FlowIDToInt(flow_id)]
      cursor.execute(query, args)
  
      requests = {}
      for req, needs_processing, responses_expected, ts in cursor.fetchall():
        if not needs_processing:
          continue
  
        request = rdf_flow_objects.FlowRequest.FromSerializedString(req)
        request.needs_processing = needs_processing
        request.nr_responses_expected = responses_expected
        request.timestamp = mysql_utils.TimestampToRDFDatetime(ts)
        requests[request.request_id] = request
  
      query = ("SELECT response, status, iterator, UNIX_TIMESTAMP(timestamp) "
               "FROM flow_responses "
               "WHERE client_id=%s AND flow_id=%s")
      cursor.execute(query, args)
  
      responses = {}
      for res, status, iterator, ts in cursor.fetchall():
        if status:
          response = rdf_flow_objects.FlowStatus.FromSerializedString(status)
        elif iterator:
          response = rdf_flow_objects.FlowIterator.FromSerializedString(iterator)
        else:
          response = rdf_flow_objects.FlowResponse.FromSerializedString(res)
        response.timestamp = mysql_utils.TimestampToRDFDatetime(ts)
        responses.setdefault(response.request_id, []).append(response)
  
      res = {}
      while next_needed_request in requests:
        req = requests[next_needed_request]
        sorted_responses = sorted(
            responses.get(next_needed_request, []), key=lambda r: r.response_id)
        res[req.request_id] = (req, sorted_responses)
        next_needed_request += 1
  
      return res
docs: Reads all requests for a flow that can be processed by the worker.