如何有效地将许多变量粘贴到sql查询中（Rshiny）

Question

我正在构建一个闪亮的应用程序，用户可以通过编辑DT:表中的选定行来更新数据库中的表。

问题是，当dt:表有很多列（比如25列）时，这个过程可能会很耗时。所以我想知道是否有一种很好且有效的方法将下面查询中的“VAL”变量与dataframe列链接起来？

下面的代码正在运行，但由于我的DT:table有60多列，我真的无法坚持这个解决方案(

selected_row <- donnees[input$dt_rows_selected,]

query <- glue_sql('UPDATE myschema.mytable SET field1= ({vals*}), field2= ({vals2*}), field3 = ({vals3*}), field4= ({vals4*}), field5= ({vals5*}) WHERE id IN ({ID_field*});',
                          vals = selected_row$column1, vals2 = selected_row$column2, vals3= selected_row$column3, vals4= selected_row$column4, vals5= selected_row$column5, ID_field= selected_row$ID, .con = pool)
    
DBI::dbExecute(pool2, query)

Answer 1

这个答案的目的有两个：

• 演示（a？）正确的动作风格。我提供了一个pg_upsert函数，在该函数中我包含了（前缀为#'#）查询完成时的样子。查询是动态形成的，因此不需要用户提供的idfields=参数以外的字段的先验知识
• 演示如何使用此功能对DT-编辑做出反应。这是一种方法，而且肯定还有其他方法来制定如何处理反应性DT。如果你有一种不同的风格来跟踪DT中的变化，那么请随意使用pg_upsert并运行它

笔记：

>

它不会在每个单元格编辑时更新数据库，更改是批量的，直到用户单击更新！按钮；更改为每个单元格上的更新是可行的，但这将是一个相对琐碎的查询，不需要更新

因为您使用的是postgres，所以目标表必须有一个或多个唯一索引（请参阅“无唯一”或“排除”约束匹配ON冲突）；我将在所述表上创建样本数据和索引；如果你不明白这意味着什么，并且你的数据没有一个清晰的“id”字段，那么就按照我所做的做：添加一个id列（本地和数据库中），该列将沿着你的实际行进行排序（如果你的数据是预先存在的，并且没有id字段，这将不起作用）

id字段必须是不可编辑的，所以DT的edable=部分禁用更改该列；我包含了一个查询（在https://stackoverflow.com/a/2213199/3358272中找到），它将以编程方式告诉您这些字段；如果没有返回任何内容，则返回到上一个项目符号并修复它

pg_upsert函数会采取一些步骤来确保事情是干净的（即检查重复的ids），但不会检查不正确的新值（DT通过class我相信会为您做一些这项工作），我假设您在发送upsert之前正在验证您需要什么；

pg_upsert的返回值是符合逻辑的，表示更新的行和我们预期的一样多；这可能过于激进，尽管我想不出比nrow（value）更好的例子来正确返回；警告买方

我在闪亮的布局中包含一个可选的"dbout"表，仅用于显示数据库数据的当前状态，每次调用pg_upsert（间接地）时都会更新；如果没有进行任何更改，它仍然会查询以显示当前状态，因此是显示测试开始条件的最佳方式；同样，它是可选的。当您删除它（您应该删除它）并且没有其他内容使用do_update（）反应性时，请更改

do_update <- eventReactive(input$upbtn, ...)
output$dbout <- renderTable({ do_update(); ... })

到

observeEvent(input$upbtn, ...)
# output$dbout <- renderTable({ do_update(); ... })

否则，从不在下游使用的reactive（.）块将永远不会触发，因此您的更新将不会发生。）

该应用程序查询数据库中的所有值（进入curdata），这可能已经在您的案例中完成了。该应用程序还可以（以编程方式）查找所需的索引。如果您提前知道这些是什么，可以随意删除为idfields提供信息的查询，直接分配它（区分大小写）。

当应用程序退出时，用户编辑的数据不会存储在本地R控制台/环境中，所有更改都存储在数据库中。我的假设是，这将被形式化为一个闪亮的服务器、RStudio Connect或类似的生产服务器，在这种情况下，“控制台”没有什么意义。如果您在开发应用程序时确实需要用户更改的数据在本地R控制台上可用，那么除了使用mydatareactive值外，在重新分配mydata$data后，您还可以覆盖curdata

这里有一个示例数据的设置。不管你有6列（如这里）还是60列，前提仍然存在。（在这之后，原始数据不被使用，这是为这个答案做准备的一个抛弃。）

# pgcon <- DBI::dbConnect(...)
set.seed(42)
origdata <- iris[sample(nrow(iris), 6),]
origdata$id <- seq_len(nrow(origdata))
# setup for this answer
DBI::dbExecute(pgcon, "drop table if exists mydata")
DBI::dbWriteTable(pgcon, "mydata", origdata)
# postgres upserts require 'unique' index on 'id'
DBI::dbExecute(pgcon, "create unique index mydata_id_idx on mydata (id)")

这是UPSERT函数本身，为了便于测试、控制台评估和类似的操作，将其分解。

#' @param value 'data.frame', values to be updated, does not need to
#'   include all columns in the database
#' @param name 'character', the table name to receive the updated
#'   values
#' @param idfields 'character', one or more id fields that are present
#'   in both the 'value' and the database table, these cannot change
#' @param con database connection object, from [DBI::dbConnect()]
#' @param verbose 'logical', be verbose about operation, default true
#' @return logical, whether 'nrow(value)' rows were affected; if an
#'   error occurred, it is messaged to the console and a `FALSE` is
#'   returned
pg_upsert <- function(value, name, idfields, con = NULL, verbose = TRUE) {
  if (verbose) message(Sys.time(), " upsert ", name, " with ", nrow(value), " rows")
  if (any(duplicated(value[idfields]))) {
    message("'value' contains duplicates in the idfields, upsert will not work")
    return(FALSE)
  }
  tmptable <- paste(c("uptemp_", name, "_", sample(1e6, size = 1)), collapse = "")
  on.exit({
    DBI::dbExecute(con, paste("drop table if exists", tmptable))
  }, add = TRUE)
  DBI::dbWriteTable(con, tmptable, value)
  cn <- colnames(value)
  quotednms <- DBI::dbQuoteIdentifier(con, cn)
  notid <- DBI::dbQuoteIdentifier(con, setdiff(cn, idfields))
  qry <- sprintf(
    "INSERT INTO %s ( %s )
     SELECT %s FROM %s
     ON CONFLICT ( %s ) DO
     UPDATE SET %s",
    name, paste(quotednms, collapse = " , "),
    paste(quotednms, collapse = " , "), tmptable,
    paste(DBI::dbQuoteIdentifier(con, idfields), collapse = " , "),
    paste(paste(notid, paste0("EXCLUDED.", notid), sep = "="), collapse = " , "))
  #'# INSERT INTO mydata ( "Sepal.Length" , "Petal.Length" )
  #'#      SELECT "Sepal.Length" , "Petal.Length" , "id" FROM mydata
  #'#      ON CONFLICT ( "id" ) DO
  #'#      UPDATE SET "Sepal.Length"=EXCLUDED."Sepal.Length" , "Petal.Length"=EXCLUDED."Petal.Length"
  # dbExecute returns the number of rows affected, this ensures we
  # return a logical "yes, all rows were updated" or "no, something
  # went wrong"
  res <- tryCatch(DBI::dbExecute(con, qry), error = function(e) e)
  if (inherits(res, "error")) {
    msg <- paste("error upserting data:", conditionMessage(res))
    message(Sys.time(), " ", msg)
    ret <- FALSE
    attr(ret, "error") <- conditionMessage(res)
  } else {
    ret <- (res == nrow(value))
    if (!ret) {
      msg <- paste("expecting", nrow(value), "rows updated, returned", res, "rows updated")
      message(Sys.time(), " ", msg)
      attr(ret, "error") <- msg
    }
  }
  ret
}

这是一个闪亮的应用程序。当您找到这个源代码时，您可以立即按更新！来获取数据库表的当前状态（同样，这只是一个选项，生产时不需要），不需要更新值来请求。

library(shiny)
library(DT)

pgcon <- DBI::dbConnect(...) # fix this incomplete expression

curdata <- DBI::dbGetQuery(pgcon, "select * from mydata order by id")
# if you don't know the idfield(s) offhand, then use this:
idfields <- DBI::dbGetQuery(pgcon, "
  select
      t.relname as table_name,
      i.relname as index_name,
      a.attname as column_name
  from
      pg_class t,
      pg_class i,
      pg_index ix,
      pg_attribute a
  where
      t.oid = ix.indrelid
      and i.oid = ix.indexrelid
      and a.attrelid = t.oid
      and a.attnum = ANY(ix.indkey)
      and t.relkind = 'r'
      and t.relname = 'mydata'
  order by
      t.relname,
      i.relname;")
idfieldnums <- which(colnames(curdata) %in% idfields$column_name)

shinyApp(
  ui = fluidPage(
    DTOutput("tbl"),
    actionButton("upbtn", "UPSERT!"),
    tableOutput("dbout")
  ),
  server = function(input, output) {

    mydata <- reactiveValues(data = curdata, changes = NULL)

    output$tbl = renderDT(
      mydata$data, options = list(lengthChange = FALSE),
      editable = list(target = "cell", disable = list(columns = idfields)))

    observeEvent(input$tbl_cell_edit, {
      mydata$data <- editData(mydata$data, input$tbl_cell_edit)
      mydata$changes <- rbind(
        if (!is.null(mydata$changes)) mydata$changes,
        input$tbl_cell_edit
      )
      # keep the most recent change to the same cell
      dupes <- rev(duplicated(mydata$changes[rev(seq(nrow(mydata$changes))),c("row","col")]))
      mydata$changes <- mydata$changes[!dupes,]
      message(Sys.time(), " pending changes: ", nrow(mydata$changes))
    })

    do_update <- eventReactive(input$upbtn, {
      if (isTRUE(nrow(mydata$changes) > 0)) {
        # always include the 'id' field(s)
        # idcol <- which(colnames(mydata$data) == "id")
        updateddata <- mydata$data[ mydata$changes$row, c(mydata$changes$col, idfieldnums) ]
        res <- pg_upsert(updateddata, "mydata", idfields = "id", con = pgcon)
        # clear the stored changes only if the upsert was successful
        if (res) mydata$changes <- mydata$changes[0,]
      }
      input$upbtn
    })

    output$dbout <- renderTable({
      do_update() # react when changes are attempted, the button is pressed
      message(Sys.time(), " query 'mydata'")
      DBI::dbGetQuery(pgcon, "select * from mydata order by id")
    })

  }
)

在行动中：

• （左）当我们开始时，我们看到原始的DT，没有数据库输出