R/connection_helpers.R

In OmopOnSpark: Using a Common Data Model on 'Spark'

# returns either "odbc" or "sparklyr"
# will give error if odbc is set up using native query
con_type <- function(con){
  if(inherits(con, "spark_connection")){
    "sparklyr"
  } else if("odbc.version" %in% names(DBI::dbGetInfo(con) |> unlist())){
    "odbc"
  } else if(!is.null(attr(con, "jConnection"))){
      "jdbc"
    } else {
    cli::cli_abort("Unkown connection type")
  }
}

validateConnection <- function(con, call = parent.frame()) {
  if(con_type(con) == "odbc"){
    validateOdbcConnection(con)
  } else if(con_type(con) == "jdbc"){
    validateJdbcConnection(con)
  } else {
    validateSparklyrConnection(con)
  }
  return(con)
}

validateOdbcConnection <- function(con){
  if(isTRUE(native_odbc_con(con))){
    cli::cli_abort("Native query via ODBC is not currently supported.")
  }
}

validateJdbcConnection <- function(con){
}

# checks if odbc connection has been set to use natice queries (TRUE) or not (FALSE)
native_odbc_con <- function(con){
  tryCatch({
    tblName <- omopgenerics::uniqueTableName()
    DBI::dbWriteTable(conn = con,
                      name = tblName,
                      value = datasets::cars |> utils::head(1))
    DBI::dbRemoveTable(conn = con,
                       name = tblName)
    FALSE
  }, error = function(e) {
    TRUE
  })
}

validateSparklyrConnection <- function(con){
if (!sparklyr::connection_is_open(con)) {
  cli::cli_abort(c(x = "{.arg con} connection is closed, please provide an open connection."), call = call)
}
}