docs: def _ReadLockAndUpdateCompletedRequests(self, request_keys, response_counts,
                                            cursor):
      """Reads, locks, and updates completed requests."""
  
      condition_template = """
        (flow_requests.client_id = %s AND
         flow_requests.flow_id = %s AND
         flow_requests.request_id = %s AND
         responses_expected = %s)"""
  
      args = []
      conditions = []
      completed_requests = {}
  
      for request_key in request_keys:
        client_id, flow_id, request_id = request_key
        if request_key in response_counts:
          conditions.append(condition_template)
          args.append(db_utils.ClientIDToInt(client_id))
          args.append(db_utils.FlowIDToInt(flow_id))
          args.append(request_id)
          args.append(response_counts[request_key])
  
      if not args:
        return completed_requests
  
      query = """
        SELECT client_id, flow_id, request_id, request
        FROM flow_requests
        WHERE ({conditions}) AND NOT needs_processing
        FOR UPDATE
      """
  
      query = query.format(conditions=" OR ".join(conditions))
      cursor.execute(query, args)
      for client_id_int, flow_id_int, request_id, request in cursor.fetchall():
        request_key = (db_utils.IntToClientID(client_id_int),
                       db_utils.IntToFlowID(flow_id_int), request_id)
        r = rdf_flow_objects.FlowRequest.FromSerializedString(request)
        completed_requests[request_key] = r
  
      query = """
      UPDATE flow_requests
      SET needs_processing = TRUE
      WHERE ({conditions}) AND NOT needs_processing
      """
      query = query.format(conditions=" OR ".join(conditions))
      cursor.execute(query, args)
  
      return completed_requests