[ideas] storage via api endpoints

[yonisidi]

Help

• I understand and agree to https://books.ropensci.org/targets/help.html.

Description

I have storage endpoints similar in nature to how targets manages aws s3 (or gcp) storage that i want to connect the pipeline to for managing the objects created by targets.

Is there a good way to leverage the infrastructure of targets to do this? or would i need to build one from scratch to mimic what class_aws, class_aws_file and class_inventory_aws do?

alternatively, could the aws infra that is built for targets be generalized to support any storage endpoint and aws would be just a case of it?

[yonisidi]

my initial attempt looks like this

#' @export
store_class_repository.api <- function(repository, store, format) {
  format <- gsub(pattern = "\\&.*$", replacement = "", x = format)
  c(
    sprintf("tar_api_%s", format),
    "tar_api",
    "tar_cloud",
    if_any(inherits(store, "tar_external"), character(0), "tar_external"),
    class(store)
  )
}

#' @export
store_assert_repository_setting.api <- function(repository) {
}

#' @export
store_produce_path.tar_api <- function(store, name, object, path_store) {
  store_produce_api_path(
    store = store,
    name = name,
    object = object,
    path_store = path_store
  )
}

store_produce_api_path <- function(store, name, object, path_store) {
  triplicate <- store$resources$api$triplicate %|||% store$resources$triplicate
  tar_assert_nonempty(triplicate)
  tar_assert_chr(triplicate)
  tar_assert_scalar(triplicate)
  tar_assert_nzchar(triplicate)

  root_prefix <- store$resources$api$prefix %|||%
    store$resources$prefix %|||%
    path_store_default()
  
  prefix <- path_objects_dir(path_store = root_prefix)
  
  tar_assert_nonempty(prefix)
  tar_assert_chr(prefix)
  tar_assert_scalar(prefix)
  
  key <- file.path(prefix, name)
  tar_assert_nzchar(key)
  triplicate <- paste0("triplicate=", triplicate)

  key <- paste0("key=", key)
  c(triplicate, key)
}

store_api_key <- function(path) {
  keyvalue_field(x = path, pattern = "^key=")
}

store_api_path_field <- function(path, pattern) {
  keyvalue_field(x = path, pattern = pattern)
}

keyvalue_field <- function(x, pattern) {
  element <- grep(pattern = pattern, x = x, value = TRUE)
  gsub(pattern = pattern, replacement = "", x = element)
}

api_base_req <- function(
    base_url = Sys.getenv('API_ROOT_URL'),
    token = Sys.getenv('API_TOKEN')
    ){
  httr2::request(base_url = base_url) |> 
    httr2::req_headers(
      "Accept" = "application/json",
      "Authorization" = sprintf('Bearer %s', token),
      .redact = 'Authorization'
    ) |>
    httr2::req_method('GET')
}

#' @export
store_read_object.tar_api <- function(store) {
  path <- store$file$path
  key <- store_api_key(path)
  scratch <- path_scratch_temp_network(pattern = basename(store_api_key(path)))
  on.exit(unlink(scratch))
  dir_create(dirname(scratch))

  api_base_req() |>
    httr2::req_url_path_append("/file/download") |>
    httr2::req_url_query(path = path) |>
    httr2::req_perform(path = scratch)
  
  store_convert_object(store, store_read_path(store, scratch))
}

#' @export
store_exist_object.tar_api <- function(store, name = NULL) {

  api_base_req() |>
    httr2::req_url_path_append("/file") |>
    httr2::req_url_query(path = dirname(store$file$path)) |>
    httr2::req_perform() |>
    httr2::resp_body_json() |>
    sapply(function(x, nm) x$name == nm, nm = name, simplify = TRUE) |>
    any()
  
}

#' @export
store_delete_objects.tar_api <- function(store) {
  
  message(
    "Can not delete objects in current api\n",
    "You need to delete them manually."
  )
  
}

#' @export
store_upload_object.tar_api <- function(store) {
  on.exit(unlink(store$file$stage, recursive = TRUE, force = TRUE))
  store_upload_object_api(store)
}

store_upload_object_api <- function(store) {

  api_base_req() |>
    httr2::req_method('POST') |>
    httr2::req_url_path_append("/file/upload") |>
    httr2::req_url_query(folder_path = path) |>
    httr2::req_body_multipart(file = curl::form_file(store$file$path)) |>
    httr2::req_perform()
  
  invisible()
  
}

#' @export
store_has_correct_hash.tar_api <- function(store) {
  hash <- store_api_hash(store)
  !is.null(hash) && identical(hash, store$file$hash)
}

store_api_hash <- function(store) {
  tar_runtime$inventories$api <- tar_runtime$inventories$api %|||%
    inventory_api_init()
  tar_runtime$inventories$api$get_cache(store = store)
}
# nocov end

#' @export
store_get_packages.tar_api <- function(store) {
  c("httr2", NextMethod())
}


# inventory ----

inventory_api_init <- function() {
  out <- inventory_api_new()
  out$reset()
  out
}

inventory_api_new <- function() {
  inventory_api_class$new()
}

inventory_api_class <- R6::R6Class(
  classname = "tar_inventory_api",
  inherit = inventory_class,
  class = FALSE,
  portable = FALSE,
  cloneable = FALSE,
  public = list(
    get_key = function(store) {
      store_api_key(store$file$path)
    },
    set_cache = function(store) {
      path <- store$file$path

      # TODO
    }
  )
)

[wlandau]

If you're willing to use content-addressable storage (CAS), then you could use the new interface that came with the latest release: https://docs.ropensci.org/targets/reference/tar_repository_cas.html. Also c.f. #1232, #1323.

[wlandau]

For tar_read() to get the envvars from tar_resources_repository_cas(), you will probably need to call tar_load_globals() first.

[yonisidi]

in the internal functions or in _targets.R?

[yonisidi]

oh, that needs to be run in the session via the command line.

I wrote a yaml i/o instead to manage the env vars in a persistent fashion instead.

[wlandau]

tar_make() uses the tar_resources_repository_cas() information set inside _targets.R, and it passes this information along to parallel workers when it runs the pipeline. tar_read(), on the other hand, runs in your interactive session (forgetting for a moment about tarchetypes::tar_quarto() etc.) and is unaware of the content inside _targets.R. tar_load_globals() is an interactive function that reads _targets.R and brings that information into your global environment.

[yonisidi]

got it. thanks for that clarification. I think for my purposes it is easier to have that info in a yaml that can support multiple storage endpoints programmatically.

if i havent said it yet... i love the cas solution, super elegant and flexible. we will make lots of use of it internally.

[yonisidi]

Followup just to see we are using as intended when packaging the functions

if I do this target’s can find exported functions of the package, similar behaviour like when in R subdir it cant find functions outside of the main functions.

targets::tar_option_set(
  repository =   tar_repository_cas(
    upload = targets.egds::wise_upload(key, path) ,
    download = targets.egds::wise_download(key, path),
    exists = targets.egds::wise_exists(key)
  ),
  resources =  tar_resources(
    repository_cas = tar_resources_repository_cas(
      envvars = c(
        MY_TOKEN = Sys.getenv('MY_TOKEN')
      )
    )
  )
)

But this works

targets::tar_option_set(
  repository =   tar_repository_cas(
    upload = function(key, path) targets.egds::wise_upload(key, path),
    download = function(key, path) targets.egds::wise_download(key, path),
    exists = function(key) targets.egds::wise_exists(key)
  ),
  resources =  tar_resources(
    repository_cas = tar_resources_repository_cas(
      envvars = c(
        MY_TOKEN = Sys.getenv('MY_TOKEN')
      )
    )
  )
)

If i try to put those wrapped functions in the package and export it ie

targets.egds::wise_repository <- function(){

tar_repository_cas(
    upload = function(key, path) targets.egds::wise_upload(key, path),
    download = function(key, path) targets.egds::wise_download(key, path),
    exists = function(key) targets.egds::wise_exists(key)
  )
}

then i get back to the same problem that it cant find functions in the package.

is this behaviour intended?

[wlandau]

It makes sense that the first example does not work, it's because those expressions are not functions. But it's odd that the last example does not work. Here is a reprex of it working on my end:

# _targets.R
library(targets)
library(tarchetypes)
declare_repository <- function() {
  tar_repository_cas(
    upload = function(key, path) {
      if (dir.exists(path)) {
        stop("This CAS repository does not support directory outputs.")
      }
      if (!file.exists("cas")) {
        dir.create("cas", recursive = TRUE)
      }
      file.rename(path, file.path("cas", key))
    },
    download = function(key, path) {
      file.copy(file.path("cas", key), path)
    },
    exists = function(key) {
      file.exists(file.path("cas", key))
    },
    list = function(keys) {
      keys[file.exists(file.path("cas", keys))]
    },
    consistent = FALSE
  )
}

write_file <- function(object) {
  writeLines(as.character(object), "file.txt")
  "file.txt"
}

list(
  tar_target(x, c(2L, 4L), repository = declare_repository()),
  tar_target(
    y,
    x,
    pattern = map(x),
    format = "qs",
    repository = declare_repository()
  ),
  tar_target(
    z,
    write_file(y),
    format = "file",
    repository = declare_repository()
  )
)

tar_make()
#> ▶ dispatched target x
#> ● completed target x [0.001 seconds, 48 bytes]
#> ▶ dispatched branch y_5ea9d15a285adf66
#> ● completed branch y_5ea9d15a285adf66 [0 seconds, 72 bytes]
#> ▶ dispatched branch y_4cd96100022a8ff3
#> ● completed branch y_4cd96100022a8ff3 [0 seconds, 72 bytes]
#> ● completed pattern y 
#> ▶ dispatched target z
#> ● completed target z [0 seconds, 4 bytes]
#> ▶ ended pipeline [0.189 seconds]

Note: I am using the development version of targets, where tar_repository_cas() has a special "list" function to remove this burden from the "exists" function.

[yonisidi]

ok, I will try that.

should this work in theory?

targets::tar_option_set(
  repository =   tar_repository_cas(
    upload = targets.egds::wise_upload,
    download = targets.egds::wise_download,
    exists = targets.egds::wise_exists
  ),
  resources =  tar_resources(
    repository_cas = tar_resources_repository_cas(
      envvars = c(
        MY_TOKEN = Sys.getenv('MY_TOKEN')
      )
    )
  )
)

[yonisidi]

so if I export this in a package

#' @export
wise_repository <- function(profile = Sys.getenv("R_CONFIG_ACTIVE")){
  withr::with_envvar(new = c("R_CONFIG_ACTIVE" = profile), {
    targets::tar_repository_cas(
        upload = function(key, path) wise_upload(key, path),
        download = function(key, path) wise_download(key, path),
        exists = function(key) wise_exists(key)
      )
    })
}

and then call

targets::tar_option_set(
  repository =   targets.egds::wise_repository(),
  resources =  tar_resources(
    repository_cas = tar_resources_repository_cas()
  )
)

# Replace the target list below with your own:
list(
  tar_target(
    name = data,
    command = tibble::tibble(x = rnorm(100), y = rnorm(100))
  ))

I get this error

r$> targets::tar_make()
- The project is out-of-sync -- use `renv::status()` for details.
▶ dispatched target data
✖ errored target data
✖ errored pipeline [0.112 seconds]
Error:
! targets::tar_make() error

── Debug target data ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
    tar_meta(data)$error
    tar_workspace(data)

── General debugging ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
    • tar_errored()
    • tar_meta(fields = any_of("error"), complete_only = TRUE)
    • tar_workspace()
    • tar_workspaces()

── How to ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
    • Debug: https://books.ropensci.org/targets/debugging.html
    • Help: https://books.ropensci.org/targets/help.html

── Last error message ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
    _store_ could not find function "wise_upload"

── Last error traceback ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
    No traceback available.

if I call targets.egds::wise_upload it works, would CMD check get upset that I am calling the package to itself like that? I haven't done that before.

[wlandau]

should this work in theory?

Yes, but if those functions are quite long, then the string generated by tar_repository_cas() will be long. That can matter because the output is string is stored in the metadata for every target that uses it. function(key, path) targets.egds::wise_upload(key, path) will give you a more compact end product.

Also, I think I see where the problems are in:

wise_repository <- function(profile = Sys.getenv("R_CONFIG_ACTIVE")){
  withr::with_envvar(new = c("R_CONFIG_ACTIVE" = profile), {
    targets::tar_repository_cas(
        upload = function(key, path) wise_upload(key, path),
        download = function(key, path) wise_download(key, path),
        exists = function(key) wise_exists(key)
      )
    })
}

First, functions wise_upload(), wise_download(), and wise_exists() need to be namespaces in the call to tar_repository_cas(). tar_repository_cas() produces a Base64-encoded string of the deparsed functions, so when they are re-parsed and evaluated, they won't necessarily remember that they came from targets.egds. Also, this is the same reason why R_CONFIG_ACTIVE will not get picked up by wise_upload() etc. For tar_make(), tar_resources_repository_cas() can provide that envvar to the CAS functions.

[yonisidi]

got it. thanks! I think I get it now. being able to choose which node is saved where is very liberating :)