R/cypher.R

In neo2R: Neo4j to R

###############################################################################@
#' Run a cypher query
#'
#' @param graph the neo4j connection
#' @param query the cypher query
#' @param parameters parameters for the cypher query.
#' @param result the way to return results. "row" will return a data frame
#' and "graph" will return a list of nodes, a list of relationships
#' and a list of paths (vectors of relationships identifiers).
#' 'source' will return the results as returned by neo4j
#' (Caution: The format is inconsistent between the two versions of the API).
#' @param arraysAsStrings if result="row" and arraysAsStrings is TRUE (default)
#' array from neo4j are converted to strings and array elements are
#' separated by eltSep.
#' @param eltSep if result="row" and arraysAsStrings is TRUE (default)
#' array from neo4j are converted to strings and array elementes are
#' separated by eltSep.
#'
#' @return The "result" of the query (invisible). See the "result" param.
#'
#' @seealso [multicypher()], [startGraph()], [prepCql()],
#' [readCql()] and [graphRequest()]
#'
#' @examples \dontrun{
#' # 2 identical queries
#' result <- cypher(
#'    graph=graph,
#'    query='match (n {value:$value}) return n',
#'    parameters=list(value="100"),
#'    result="graph"
#' )
#' result <- cypher(
#'    graph=graph,
#'    query='match (n {value:"100"}) return n',
#'    result="graph"
#' )
#' }
#'
#' @export
#'
cypher <- function(
  graph,
  query,
  parameters = NULL,
  result = c("row", "graph", "source"),
  arraysAsStrings = TRUE,
  eltSep = " || "
) {
  result <- match.arg(result)
  endpoint <- graph$cypher_endpoint
  api <- if (is.null(graph$api)) "tx" else graph$api

  if (api == "v2") {
    postText <- list(statement = query)
    if (!is.null(parameters)) postText$parameters <- parameters
  } else {
    postText <- list(
      statements = list(list(
        statement = query,
        resultDataContents = list(ifelse(result == "source", "graph", result))
      ))
    )
    if (!is.null(parameters)) {
      postText$statements[[1]]$parameters <- parameters
    }
  }

  raw_result <- graphRequest(
    graph = graph,
    endpoint = endpoint,
    customrequest = "POST",
    postText = postText
  )$result
  errors <- raw_result$errors
  if (length(errors) > 0) {
    devnull <- lapply(errors, lapply, message)
    stop("neo4j error")
  }
  if (api == "v2") {
    if (result == "source") {
      toRet <- raw_result$data
    } else {
      if (result == "row") {
        toRet <- process_row_v2(
          raw_result$data,
          arraysAsStrings = arraysAsStrings,
          eltSep = eltSep
        )
      } else {
        toRet <- process_graph_v2(raw_result$data)
      }
    }
  } else {
    if (result == "source") {
      toRet <- raw_result$results[[1]]
    } else {
      if (result == "row") {
        toRet <- process_row(
          raw_result$results[[1]],
          arraysAsStrings = arraysAsStrings,
          eltSep = eltSep
        )
      }
      if (result == "graph") {
        toRet <- process_graph(raw_result$results[[1]]$data)
      }
    }
  }
  invisible(toRet)
}

###############################################################################@
#' Run a multiple cypher queriers
#'
#' @param graph the neo4j connection
#' @param queries queries to submit. It can be either a character vector
#' for which each element corresponds to a cypher query. Or it can be
#' a list of lists with the following slots:
#' - **query** (mandatory): A single character corresponding to the cypher
#' query.
#' - **parameters** (optional): A set of parameters specific for
#' this query. If not provided, the *parameters* parameter of the function is
#' used  (see below).
#' - **result** (optional): The specific way to return the results of
#' this query. If not provided, the *result* parameter of the function is
#' used  (see below).
#' @param parameters default parameters for the cypher queries.
#' @param result default way to return results. "row" will return a data frame
#' and "graph" will return a list of nodes, a list of relationships
#' and a list of paths (vectors of relationships identifiers).
#' 'source' will return the results as returned by neo4j
#' (Caution: The format is inconsistent between the two versions of the API).
#' @param arraysAsStrings if result="row" and arraysAsStrings is TRUE (default)
#' array from neo4j are converted to strings and array elements are
#' separated by eltSep.
#' @param eltSep if result="row" and arraysAsStrings is TRUE (default)
#' array from neo4j are converted to strings and array elementes are
#' separated by eltSep.
#'
#' @return A list of "result" of the queries (invisible).
#' See the "result" param.
#'
#' @seealso [cypher()], [startGraph()], [prepCql()],
#' [readCql()] and [graphRequest()]
#'
#' @examples \dontrun{
#' result <- multicypher(
#'    graph,
#'    queries=list(
#'       q1="match (n) return n.value limit 5",
#'       q2=list(
#'          query="match (f {value:$val})-[r]->(t) return f, r, t limit 5",
#'          result="graph",
#'          parameters=list(val=100)
#'       )
#'    )
#' )
#' }
#'
#' @export
#'
multicypher <- function(
  graph,
  queries,
  parameters = NULL,
  result = c("row", "graph", "source"),
  arraysAsStrings = TRUE,
  eltSep = " || "
) {
  result <- match.arg(result)
  endpoint <- graph$cypher_endpoint
  api <- if (is.null(graph$api)) "tx" else graph$api
  qnames <- names(queries)

  ## Query API v2: one request per query ----
  if (api == "v2") {
    toRet <- lapply(
      unname(queries),
      function(query) {
        if (is.character(query)) {
          cypher(
            graph,
            query = query,
            parameters = parameters,
            result = result,
            arraysAsStrings = arraysAsStrings,
            eltSep = eltSep
          )
        } else {
          if (!"query" %in% names(query)) {
            stop('Parameterized query should have a "query" slot')
          }
          qparams <- if (!is.null(query$parameters)) {
            query$parameters
          } else {
            parameters
          }
          qresult <- if (!is.null(query$result)) {
            match.arg(query$result, c("row", "graph", "source"))
          } else {
            result
          }
          cypher(
            graph,
            query = query$query,
            parameters = qparams,
            result = qresult,
            arraysAsStrings = arraysAsStrings,
            eltSep = eltSep
          )
        }
      }
    )
    names(toRet) <- qnames
    return(invisible(toRet))
  }

  statements <- lapply(
    unname(queries),
    function(query) {
      if (!is.character(query) && !is.list(query)) {
        stop("Each query should be character or a list")
      }
      if (is.character(query)) {
        toRet <- list(statement = query)
      }
      if (is.list(query)) {
        if (!"query" %in% names(query)) {
          stop('Parameterized query should have a "query" slot')
        }
        if (!is.character(query$query)) {
          stop('The "query" slot should be a character')
        }
        toRet <- list(statement = query$query)
        if ("result" %in% names(query)) {
          if (!query$result %in% c("row", "graph", "source")) {
            stop('The "result" slot should be "row", "graph" or "source"')
          }
          toRet$resultDataContents <- list(ifelse(
            query$result == "source",
            "graph",
            query$result
          ))
          toRet$result <- list(query$result)
        }
        if (!is.null(query$parameters)) {
          toRet$parameters <- query$parameters
        }
      }
      if (is.null(toRet$resultDataContents)) {
        toRet$resultDataContents <- list(ifelse(
          result == "source",
          "graph",
          result
        ))
        toRet$result <- list(result)
      }
      if (is.null(toRet$parameters) && !is.null(parameters)) {
        toRet$parameters <- parameters
      }
      return(toRet)
    }
  )
  postText <- list(statements = statements)

  results <- graphRequest(
    graph = graph,
    endpoint = endpoint,
    customrequest = "POST",
    postText = postText
  )$result
  errors <- results$errors
  if (length(errors) > 0) {
    devnull <- lapply(errors, lapply, message)
    stop("neo4j error")
  }

  toRet <- lapply(
    1:length(results$results),
    function(i) {
      results <- results$results[[i]]
      if (statements[[i]]$result == "source") {
        return(results)
      }
      if (statements[[i]]$result == "row") {
        return(process_row(
          results,
          arraysAsStrings = arraysAsStrings,
          eltSep = eltSep
        ))
      }
      if (statements[[i]]$result == "graph") {
        return(process_graph(results$d))
      }
    }
  )
  names(toRet) <- qnames
  invisible(toRet)
}


###############################################################################@
## Helpers ----
process_row <- function(results, arraysAsStrings, eltSep) {
  if (length(results$data) == 0) {
    toRet <- NULL
  } else {
    columns <- do.call(c, results$columns)
    toRet <- do.call(rbind, lapply(results$data, function(x) x$row))
    toRet[sapply(toRet, is.null)] <- NA
    toRet <- data.frame(toRet, stringsAsFactors = FALSE)
    if (all(sapply(toRet, class) == "list")) {
      for (i in 1:ncol(toRet)) {
        if (max(unlist(sapply(toRet[[i]], length))) == 1) {
          toRet[, i] <- unlist(toRet[, i])
        } else {
          if (arraysAsStrings) {
            toRet[, i] <- unlist(lapply(
              toRet[, i],
              paste,
              collapse = eltSep
            ))
          }
        }
      }
    }
    colnames(toRet) <- columns
  }
  return(toRet)
}

process_row_v2 <- function(data, arraysAsStrings, eltSep) {
  if (is.null(data) || length(data$values) == 0) {
    return(NULL)
  }
  process_row(
    list(
      columns = data$fields,
      data = lapply(data$values, function(row) list(row = row))
    ),
    arraysAsStrings,
    eltSep
  )
}

process_graph <- function(d) {
  if (is.null(d) || length(d) == 0) {
    return(NULL)
  }
  nodes <- unique(do.call(c, lapply(d, function(x) x$graph$nodes)))
  names(nodes) <- unlist(lapply(nodes, function(n) n$id))
  relationships <- unique(do.call(
    c,
    lapply(
      d,
      function(x) x$graph$relationships
    )
  ))
  names(relationships) <- unlist(lapply(relationships, function(n) n$id))
  p <- lapply(
    d,
    function(x) {
      unique(unlist(lapply(x$graph$relationships, function(y) y$id)))
    }
  )
  p <- p[which(!unlist(lapply(p, is.null)))]
  toRet <- list(nodes = nodes, relationships = relationships, paths = p)
}

process_graph_v2 <- function(d) {
  if (is.null(d) || length(d$values) == 0) {
    return(NULL)
  }
  is_scalar <- function(v) {
    !is.list(v)
  }
  is_node <- function(v) {
    is.list(v) &&
      !is.null(v$elementId) &&
      is.null(v$startNodeElementId) &&
      is.null(v$type)
  }
  is_rel <- function(v) {
    is.list(v) &&
      !is.null(v$elementId) &&
      !is.null(v$type) &&
      !is.null(v$startNodeElementId)
  }
  nodes <- list()
  relationships <- list()
  p <- list()
  for (row in d$values) {
    for (val in row) {
      if (is_node(val)) {
        nodes[[val$elementId]] <- val
      }
      if (is_rel(val)) {
        relationships[[val$elementId]] <- val
      }
      if (!is_scalar(val)) {
        row_rel_ids <- character(0)
        for (sval in val) {
          if (is_node(sval)) {
            nodes[[sval$elementId]] <- sval
          }
          if (is_rel(sval)) {
            relationships[[sval$elementId]] <- sval
            row_rel_ids <- c(row_rel_ids, sval$elementId)
          }
        }
        if (length(row_rel_ids) > 0) {
          p[[length(p) + 1]] <- row_rel_ids
        }
      }
    }
  }
  nodes <- unique(nodes)
  names(nodes) <- unlist(lapply(nodes, function(n) n$elementId))
  relationships <- unique(relationships)
  names(relationships) <- unlist(lapply(relationships, function(n) n$elementId))
  list(
    nodes = nodes,
    relationships = relationships,
    paths = unique(p)
  )
}