R/td.R

In teramonagi/tdclientr: Treasure Data API library for R

#' Create Treasure Data client
#'
#' @param api_key API key for Treasure Data. You can get this from "API Keys" on https://console.treasuredata.com/users/current
#' @param database Database on Treasure Data to connect.
#' @export
td <- function(api_key=NULL, database=NULL)
{
  if(is.null(api_key)){
    if(exists("TD_API_KEY")){
      api_key <- TD_API_KEY
    }
    if(Sys.getenv("TD_API_KEY") != ""){
      api_key <- Sys.getenv("TD_API_KEY")
    }
    if(is.null(api_key)){stop("api_key must be given when there no TD_API_KEY (environment or usual R) variable")}
  }
  structure(list(key=api_key, db=database), class="td")
}

#' @export
is.td <- function(x){inherits(x, "td")}

stop_when_not_td <- function(client)
{
  if(!is.td(client)){
    stop("Error: client is not td object")
  }
}

select_db <- function(db1, db2)
{
  ifelse(is.null(db2), db1, db2)
}

stop_when_not_specify_db <- function(db1, db2)
{
  if(is.null(db1) && is.null(db2)){
    stop("You have to specify database as argument or client object(db field)")
  }
}

#' Return a list of your jobs.
#'
#' @param client Treasure Data client generated by td function.
#' @return data.frame containing an array of the jobs.
#' @export
list_job <- function(client)
{
  stop_when_not_td(client)

  url <- paste0(url_api, "job/list")
  response <- get(client$key, url)
  jsonlite::fromJSON(content(response, "text"))$jobs %>%
    dplyr::mutate_each(funs(to_posixct), created_at, updated_at, start_at, end_at)
}

#' Returns a list of your tables.
#'
#' @param client Treasure Data client generated by td function.
#' @param database(optional): name of the database
#' @export
list_table <- function(client, database=NULL)
{
  stop_when_not_td(client)
  stop_when_not_specify_db(client$db, database)

  db <- select_db(client$db, database)
  url <- paste0(url_api, "table/list/", db)
  response <- get(client$key, url)
  jsonlite::fromJSON(content(response, "text"))$tables %>%
    dplyr::mutate_each(funs(to_posixct), counter_updated_at, last_log_timestamp, created_at, updated_at)
}


#' Swaps the contents of two tables.
#'
#' @param client Treasure Data client generated by td function.
#' @param table1: table name (before)
#' @param table2: table name (after)
#' @param database(optional): name of the database
#' @export
table_swap <- function(client, table1, table2, database=NULL)
{

}

#' Issue queries
#'
#' @param client Treasure Data client generated by td function.
#' @param query: query string
#' @param database(optional): name of the database
#' @param type(optional): the job type (‘hive’ or ‘presto’)
#' @param priority(optional): priority of the job. -2 (VERY LOW) to 2 (VERY HIGH). The default is 0 (NORMAL).
#' @export
issue_job <- function(client, query, database=NULL, type="presto", priority=0, wait=FALSE)
{
  stop_when_not_td(client)
  stop_when_not_specify_db(client$db, database)

  db <- select_db(client$db, database)
  url <- paste0(url_api, sprintf("job/issue/%s/%s", type, db))
  response <- post(client$key, url, list(query=query, priority=priority))
  df <- as.data.frame(jsonlite::fromJSON(content(response, "text")))
  td_job(client, df$job_id, db)
}