R/mongo.R

In EntirelyDS/edsrutils: Entirely Data Science R utility package

# Get Mongo Connection Strings
#' @export
mongo_connection_strings <- function(config, dbs){
  conn_str <- list()
  i <- 1
  for (db in dbs){
    if (
      'database' %in% names(config) &&
        'mongo' %in% names(config$database) &&
          db %in% names(config$database$mongo)
    ){
      conn_str[i] <- config$database$mongo[[db]]$uri
    }else{
      conn_str[i] <- NA
    }
    names(conn_str)[i] <- db
    i <- i + 1
  }
  return(conn_str)

}


#' @export
mongo_date <- function(x){

  if(!('POSIXct' %in% class(x))) x <- as.POSIXct(x)

  return(list('$date' = floor(unclass(x) * 1000)))

}

#' @export
df_to_mongo_update <- function(df, add_set = TRUE){
  ustr <- jsonlite::toJSON(df)
  ustr <- stringr::str_sub(ustr, 2, stringr::str_length(ustr) - 1)
  if (add_set){
    ustr <- stringr::str_c('{"$set":',ustr,'}')
  }
  return(ustr)
}

#' @export
run_mongo_file_aggregation <- function(m, qfile, ...){

  assertive.files::assert_all_are_readable_files(qfile)
  qstr <- stringr::str_c(readLines(qfile), collapse = '\n')
  dots <- list(...)
  dots <- dots[!sapply(dots, is.na)]
  call_args <- list(qstr)
  if(length(dots) > 0){
    call_args <- c(call_args, dots)
  }
  qstr <- do.call('sprintf', call_args)
  out_df <- m$aggregate(qstr)

  return(out_df)
}