# BACKLOG:
# - can only return so many objects at once, how to paginate via UI?
# - impacts experiments, model registry, etc.
# nocov start
brickster_actions <- (host) {
(
`Open Workspace` = (
icon = "",
callback = () {
::(glue::glue("https://{host}"))
}
)
)
}
# nocov end
get_id_from_panel_name <- (x) {
( = ".*\\((.*)\\).*", replacement = "\\1", x = x)
}
get_model_version_from_string <- (x) {
(( = "(\\d+).*", replacement = "\\1", x = x))
}
readable_time <- (x) {
<- (
x = x/1000,
origin = "1970-01-01",
tz = "UTC"
)
()
}
get_catalogs <- (host, token) {
catalogs <- (host = host, token = token)
(
= purrr::map_chr(catalogs, "name"),
= "catalog"
)
if ((catalogs) > 0) {
(
= purrr::map_chr(catalogs, "name"),
= "catalog"
)
} {
( = , = )
}
}
get_schemas <- (catalog, host, token) {
schemas <- (
catalog = catalog,
host = host,
token = token
)
if ((schemas) > 0) {
(
= purrr::map_chr(schemas, "name"),
= "schema"
)
} {
( = , = )
}
}
get_tables <- (catalog, schema, host, token) {
tables <- (
catalog = catalog,
schema = schema,
host = host,
token = token
)
if ((tables) > 0) {
(
= purrr::map_chr(tables, "name"),
= "table"
)
} {
( = , = )
}
}
get_uc_models <- (catalog, schema, host, token) {
models <- db_uc_models_list(
catalog = catalog,
schema = schema,
host = host,
token = token
)
if ((models) > 0) {
(
= purrr::map_chr(models, "name"),
= "model"
)
} {
( = , = )
}
}
get_uc_model <- (catalog, schema, model, host, token) {
model <- db_uc_models_get(
catalog = catalog,
schema = schema,
model = model,
host = host,
token = token
)
info <- (
"name" = model$,
"owner" = model$owner,
"created at" = readable_time(model$created_at),
"created by" = model$created_by,
"last updated" = readable_time(model$updated_at),
"updated by" = model$updated_by,
"id" = model$id
)
(
= (info),
= ((info))
)
}
get_uc_model_versions <- (catalog, schema, model, host, token,
= ) {
# if version is NULL get all, otherwise specific versions
versions <- db_uc_model_versions_get(
catalog,
schema,
model,
host = host,
token = token
)[1]]
# get model info again to get the aliases
model_info <- db_uc_models_get(catalog, schema, model, host, token)
aliases <- purrr::map(
model_info$aliases, ~{
(.x$version_num, .x$alias_name)
}) |>
()
version_names <- purrr::map_chr(versions, (x) {
if (x$ %in% aliases) {
alias_values <- (aliases[x$ %in% aliases])
alias_part <- ("@", alias_values, collapse = ", ")
(x$, " (", alias_part, ")")
} {
x$
}
})
if (()) {
res <- (
= version_names,
= "version"
)
} {
version_meta <- versions[which(purrr::map_vec(versions, "version") == )]]
info <- (
"created at" = readable_time(version_meta$created_at),
"created by" = version_meta$created_by,
"last updated" = readable_time(version_meta$updated_at),
"updated by" = version_meta$updated_by,
"run id" = version_meta$run_id,
"run workspace id" = version_meta$run_workspace_id,
"source" = version_meta$,
"status" = version_meta$status,
"id" = version_meta$id
)
res <- (
= (info),
= ((info))
)
}
res
}
get_uc_functions <- (catalog, schema, host, token) {
funcs <- db_uc_funcs_list(
catalog = catalog,
schema = schema,
host = host,
token = token
)
if ((funcs) > 0) {
(
= purrr::map_chr(funcs, "name"),
= "func"
)
} {
( = , = )
}
}
get_uc_function <- (catalog, schema, func, host, token) {
func <- db_uc_funcs_get(
catalog = catalog,
schema = schema,
func = func,
host = host,
token = token
)
info <- (
"name" = func$,
"date type" = func$data_type,
"full data type" = func$full_data_type,
"created at" = readable_time(func$created_at),
"created by" = func$created_by,
"last updated" = readable_time(func$updated_at),
"updated by" = func$updated_by,
"id" = func$function_id
)
(
= (info),
= ((info))
)
}
get_uc_volumes <- (catalog, schema, host, token) {
volumes <- (
catalog = catalog,
schema = schema,
host = host,
token = token
)
if ((volumes) > 0) {
(
= purrr::map_chr(volumes, "name"),
= "volume"
)
} {
( = , = )
}
}
get_uc_volume <- (catalog, schema, host, volume, token) {
volumes <- (
catalog = catalog,
schema = schema,
host = host,
token = token
)
volume <- purrr::keep(volumes, ~.x$ == volume)[1]]
info <- (
"name" = volume$,
"volume type" = volume$volume_type,
"storage location" = volume$storage_location,
"created at" = readable_time(volume$created_at),
"created by" = volume$created_by,
"last updated" = readable_time(volume$updated_at),
"updated by" = volume$updated_by,
"id" = volume$volume_id
)
(
= (info),
= ((info))
)
}
get_schema_objects <- (catalog, schema, host, token) {
<- ()
$tables <- get_tables(catalog, schema, host, token)
$volumes <- get_uc_volumes(catalog, schema, host, token)
$models <- get_uc_models(catalog, schema, host, token)
$funcs <- get_uc_functions(catalog, schema, host, token)
# how many objects of each type exist
# only show when objects exist within
sizes <- purrr::map_int(, ) |>
purrr::keep(~.x > 0) |>
purrr::imap_chr(~ glue::glue("{.y} ({.x})"))
(
= (sizes),
= (sizes)
)
}
get_table_data <- (catalog, schema, , host, token, metadata = ) {
# if metadata is TRUE then return metadata, otherwise columns
tbl <- (
catalog = catalog,
schema = schema,
= ,
omit_columns = ,
omit_properties = ,
omit_username = ,
host = host,
token = token,
)
# TODO: handle edge case errors?
if (metadata) {
if (tbl$table_type == "VIEW") {
info <- (
"table type" = tbl$table_type,
"view definition" = tbl$view_definition,
"full name" = tbl$full_name,
"owner" = tbl$owner,
"created at" = readable_time(tbl$created_at),
"created by" = tbl$created_by,
"updated at" = readable_time(tbl$updated_at),
"updated by" = tbl$updated_by
)
} if (tbl$data_source_format == "DELTASHARING") {
info <- (
"table type" = tbl$table_type,
"data source format" = tbl$data_source_format,
"full name" = tbl$full_name,
"owner" = tbl$owner,
"storage location" = tbl$storage_location,
"created at" = readable_time(tbl$created_at),
"created by" = tbl$created_by,
"updated at" = readable_time(tbl$updated_at),
"updated by" = tbl$updated_by
)
} if (tbl$data_source_format == "VECTOR_INDEX_FORMAT") {
info <- (
"table type" = tbl$table_type,
"data source format" = tbl$data_source_format,
"full name" = tbl$full_name,
"owner" = tbl$owner,
"endpoint name" = tbl$properties$endpoint_name,
"endpoint type" = tbl$properties$endpoint_type,
"primary key" = tbl$properties$primary_key,
"created at" = readable_time(tbl$created_at),
"created by" = tbl$created_by,
"updated at" = readable_time(tbl$updated_at),
"updated by" = tbl$updated_by
)
} if (tbl$data_source_format == "TABLE") {
info <- (
"table type" = tbl$table_type,
"data source format" = tbl$data_source_format,
"full name" = tbl$full_name,
"owner" = tbl$owner,
"storage location" = tbl$storage_location,
"created at" = readable_time(tbl$created_at),
"created by" = tbl$created_by,
"updated at" = readable_time(tbl$updated_at),
"updated by" = tbl$updated_by,
"last commit at" = readable_time((tbl$properties$delta.lastCommitTimestamp)),
"min reader version" = tbl$properties$delta.minReaderVersion,
"min writer version" = tbl$properties$delta.minWriterVersion
)
} {
info <- (
"table type" = tbl$table_type,
"data source format" = tbl$data_source_format,
"full name" = tbl$full_name,
"owner" = tbl$owner,
"created at" = readable_time(tbl$created_at),
"created by" = tbl$created_by,
"updated at" = readable_time(tbl$updated_at),
"updated by" = tbl$updated_by
)
}
} {
info <- purrr::map_chr(tbl$columns, (x) {
(x$type_name, " (nullable: ", x$nullable, ")")
})
(info) <- purrr::map_chr(tbl$columns, "name")
}
(
= (info),
= ((info))
)
}
get_experiments <- (host, token) {
experiments <- db_experiments_list(host = host, token = token)
exp_names <- purrr::map_chr(experiments, "name")
exp_ids <- purrr::map_chr(experiments, "experiment_id")
(
= ((".*\\/(.*)", "\\1", exp_names), " (", exp_ids, ")"),
= "experiment"
)
}
get_experiment <- (id, host, token) {
id <- get_id_from_panel_name(x = id)
<- db_experiments_get(id = id, host = host, token = token)
info <- (
"name" = $,
"experiment id" = $experiment_id,
"artifact location" = $artifact_location,
"user id" = $lifecycle_stage,
"created at" = readable_time($creation_time),
"last updated" = readable_time($last_update_time)
)
(
= (info),
= ((info))
)
}
get_models <- (host, token) {
models <- db_mlflow_registered_models_list(
max_results = 1000,
host = host,
token = token
)
models <- models$registered_models
(
= purrr::map_chr(models, "name"),
= "model"
)
}
get_model_metadata <- (id, host, token) {
model <- (
= id,
host = host,
token = token
)
info <- (
"name" = model$,
"latest version" = model$latest_versions[1]]$,
"user id" = model$user_id,
"created at" = readable_time(model$creation_timestamp),
"last updated" = readable_time(model$last_updated_timestamp),
"permissions" = model$permission_level,
"id" = model$id
)
(
= (info),
= ((info))
)
}
get_model_versions <- (id, host, token, = ) {
# if version is NULL get all, otherwise specific versions
versions <- db_mlflow_registered_models_search_versions(
= id,
host = host,
token = token
)[1]]
version_names <- purrr::map_chr(versions, (x) {
if (x$current_stage == "None") {
x$
} {
(x$, " (", x$current_stage, ")")
}
})
if (()) {
res <- (
= version_names,
= "version"
)
} {
version_meta <- versions[which(version_names == )]]
info <- (
"current stage" = version_meta$current_stage,
"created at" = readable_time(version_meta$creation_timestamp),
"last updated" = readable_time(version_meta$last_updated_timestamp),
"user id" = version_meta$user_id,
"source" = version_meta$,
# can be missing, not sure worth including for now
# "run id" = version_meta$run_id,
"status" = version_meta$status
)
res <- (
= (info),
= ((info))
)
}
res
}
get_clusters <- (host, token) {
clusters <- (host = host, token = token)
purrr::map_dfr(clusters, (x) {
(
= (glue::glue("[{x$state}] {x$cluster_name} ({x$cluster_id})")),
= "cluster"
)
})
}
get_cluster <- (id, host, token) {
x <- (id, host, token)
info <- (
"name" = x$cluster_name,
"id" = id,
"creator" = x$creator_user_name,
"runtime" = x$spark_version,
"worker node type" = x$node_type_id,
"driver node type" = x$driver_node_type_id,
"autotermination minutes" = x$autotermination_minutes,
"start time" = readable_time(x$start_time),
"# workers" = ((x$num_workers), 0L, x$num_workers),
"cores" = ((x$cluster_cores), 0L, x$cluster_cores),
"memory (mb)" = ((x$cluster_memory_mb), 0L, x$cluster_memory_mb)
)
(
= (info),
= ((info))
)
}
get_warehouses <- (host, token) {
warehouses <- (host = host, token = token)
purrr::map_dfr(warehouses, (x) {
(
= (glue::glue("[{x$state}] {x$name} ({x$id})")),
= "warehouse"
)
})
}
get_warehouse <- (id, host, token) {
x <- (id, host, token)
info <- (
"name" = x$,
"id" = id,
"creator" = x$creator_name,
"channel" = dplyr::coalesce(x$channel$, ),
"serverless" = x$enable_serverless_compute,
"size" = x$cluster_size,
"# clusters" = x$num_clusters,
"clusters (min)" = x$min_num_clusters,
"clusters (max)" = x$max_num_clusters,
"autostop minutes" = x$auto_stop_mins
)
(
= (info),
= ((info))
)
}
# nocov start
list_objects <- (host, token,
= ,
workspace = ,
clusters = ,
warehouses = ,
metastore = ,
catalog = ,
schema = ,
tables = ,
= ,
volumes = ,
funcs = ,
models = ,
modelregistry = ,
model = ,
versions = ,
columns = ,
experiments = ,
) {
# uc metastore
if (!(metastore)) {
if (!(schema)) {
if (!(volumes)) {
<- get_uc_volumes(catalog, schema, host, token)
()
}
if (!(funcs)) {
<- get_uc_functions(catalog, schema, host, token)
()
}
if (!(models)) {
if (!(versions)) {
<- get_uc_model_versions(catalog, schema, model, host, token)
()
}
if (!(model)) {
<- (
= ("metadata", "versions"),
= ("metadata", "versions")
)
()
}
<- get_uc_models(catalog, schema, host, token)
()
}
if (!(tables)) {
if (!()) {
<- (
= ("metadata", "columns"),
= ("metadata", "columns")
)
()
}
<- get_tables(catalog, schema, host, token)
()
}
<- get_schema_objects(catalog, schema, host, token)
()
}
if (!(catalog)) {
<- get_schemas(catalog, host, token)
()
}
# catch all, return catalogs
<- get_catalogs(host, token)
()
}
# experiments
if (!(experiments)) {
<- get_experiments(host, token)
()
}
# model registry
if (!(modelregistry)) {
if (!(versions)) {
<- get_model_versions(id = model, host, token)
()
}
if (!(model)) {
<- (
= ("metadata", "versions"),
= ("metadata", "versions")
)
()
}
# catch all to return models
<- get_models(host, token)
()
}
# clusters
if (!(clusters)) {
<- get_clusters(host, token)
()
}
# warehouses
if (!(warehouses)) {
<- get_warehouses(host, token)
()
}
# baseline view (static)
# first we should determine if workspace is single-tenant (no SQL/uc)
# check if sql endpoint fails
sql_active <- (
expr = {
(host, token)
},
error = (e)
)
# check if UC catalogs endpoint fails
uc_active <- (
expr = {
(host, token)
},
error = (e)
)
info <- (
"Catalog" = "metastore",
"Model Registry" = "modelregistry",
"Experiments" = "experiments",
"Clusters" = "clusters",
"SQL Warehouses" = "warehouses"
)
if (!sql_active) {
info"SQL Warehouses"]] <-
}
if (!uc_active) {
info"Data"]] <-
}
(
= (info),
= ((info))
)
}
list_columns <- (host, token, path = "", ) {
<- ()
leaf <- [length()]
leaf_type <- (leaf)
# for clusters and warehouses things are simple
if (leaf_type == "cluster") {
info <- get_cluster(id = get_id_from_panel_name(leaf), host = host, token = token)
(info)
}
if (leaf_type == "warehouse") {
info <- get_warehouse(id = get_id_from_panel_name(leaf), host = host, token = token)
(info)
}
if (!($modelregistry) && "model" %in% ()) {
if (leaf_type == "metadata") {
info <- get_model_metadata(id = $model, host = host, token = token)
} if (leaf_type == "version") {
info <- get_model_versions(
id = $model,
= leaf$,
host = host,
token = token
)
}
}
if (!($catalog) && "model" %in% ()) {
if (leaf_type == "metadata") {
info <- get_uc_model(
catalog = "catalog"]],
schema = "schema"]],
model = "model"]],
host = host,
token = token
)
} if (leaf_type == "version") {
info <- get_uc_model_versions(
catalog = "catalog"]],
schema = "schema"]],
model = "model"]],
= get_model_version_from_string(leaf$),
host = host,
token = token
)
}
}
if ("table" %in% ()) {
info <- get_table_data(
catalog = $catalog,
schema = $schema,
= $,
host = host,
token = token,
metadata = leaf_type == "metadata"
)
}
if (leaf_type == "func") {
info <- get_uc_function(
catalog = $catalog,
schema = $schema,
func = $func,
host = host,
token = token
)
}
if (leaf_type == "volume") {
info <- get_uc_volume(
catalog = $catalog,
schema = $schema,
volume = leaf$volume,
host = host,
token = token
)
}
if (leaf_type == "experiment") {
info <- get_experiment(
id = leaf$experiment,
host = host,
token = token
)
}
info
}
preview_object <- (host, token, rowLimit,
path = "",
cluster = ,
warehouse = ,
= ,
model = ,
= ,
experiment = ,
catalog = ,
schema = ,
= ,
func = ,
volume = ,
) {
ws_id <- ()
if (!(catalog)) {
if (!(catalog) & !(schema) & !(func)) {
path <- (("functions", catalog, schema, func), collapse = "/")
} if (!(catalog) & !(schema) & !(model) & !()) {
<- get_model_version_from_string()
path <- (("models", catalog, schema, model, "version", ), collapse = "/")
} if (!(catalog) & !(schema) & !(model)) {
path <- (("models", catalog, schema, model), collapse = "/")
} if (!(catalog) & !(schema) & !(volume)) {
path <- (("volumes", catalog, schema, volume), collapse = "/")
} if (!(catalog) & !(schema) & !()) {
path <- ((catalog, schema, ), collapse = "/")
} if (!(catalog) & !(schema)) {
path <- ((catalog, schema), collapse = "/")
} {
path <- catalog
}
<- glue::glue("https://{host}/explore/data/{path}?o={ws_id}")
(::())
}
# version of model
if (!() && !(model)) {
<- ("(\\d+) .*", "\\1", )
<- glue::glue("https://{host}/?o={ws_id}#mlflow/models/{model}/versions/{version}")
(::())
}
# model
if (() && !(model)) {
<- glue::glue("https://{host}/ml/models/{model}?o={ws_id}")
(::())
}
# experiment
if (!(experiment)) {
id <- get_id_from_panel_name(experiment)
<- glue::glue("https://{host}/?o={ws_id}#mlflow/experiments/{id}")
(::())
}
if (!(cluster)) {
id <- get_id_from_panel_name(cluster)
<- glue::glue("https://{host}/?o={ws_id}#setting/clusters/{id}/configuration")
(::())
}
if (!(warehouse)) {
id <- get_id_from_panel_name(warehouse)
<- glue::glue("https://{host}/sql/warehouses/{id}")
(::())
}
}
# nocov end
#' Connect to Databricks Workspace
#'
#' @inheritParams auth_params
#' @param name Desired name to assign the connection
#'
#' @export
#'
#' @examples
#' \dontrun{
#' open_workspace(host = db_host(), token = db_token, name = "MyWorkspace")
#' }
#' @importFrom glue glue
<- (host = (), token = (), = ) {
# nocov start
observer <- ("connectionObserver")
if (!(observer)) {
connection_string <- glue::glue(
"library(brickster)\nopen_workspace(host = db_host(), token = db_token())
")
display_name <- if (!()) host
observer$connectionOpened(
= "workspace",
host = host,
displayName = display_name,
icon = ("icons", "logo.png", package = "brickster"),
connectCode = connection_string,
disconnect = () {
(host)
},
listObjectTypes = () {
list_objects_types()
},
listObjects = ( = "root", ) {
<- ()
<- list_objects(
host,
token,
= "files"]],
warehouses = "warehouses"]],
clusters = "clusters"]],
metastore = "metastore"]],
catalog = "catalog"]],
schema = "schema"]],
= "table"]],
tables = "tables"]],
volume = "volume"]],
volumes = "volumes"]],
models = "models"]],
func = "func"]],
funcs = "funcs"]],
columns = "columns"]],
experiments = "experiments"]],
modelregistry = "modelregistry"]],
model = "model"]],
versions = "versions"]]
)
()
},
listColumns = () {
columns <- list_columns(host = host, token = token, )
(columns)
},
previewObject = (rowLimit, ) {
preview_object(host = host, token = token, rowLimit = rowLimit, )
},
actions = brickster_actions(host),
connectionObject = host
)
}
# nocov end
}
#' Close Databricks Workspace Connection
#'
#' @inheritParams auth_params
#'
#' @export
#'
#' @examples
#' \dontrun{
#' close_workspace(host = db_host())
#' }
<- (host = ()) {
observer <- ("connectionObserver")
if (!(observer)) {
observer$connectionClosed( = "workspace", host = host)
}
}
# nocov start
list_objects_types <- () {
(
workspace = (contains = (
clusters = (
icon = ("icons", "compute.png", package = "brickster"),
contains = (
cluster = (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
)
)),
metastore = (
icon = ("icons", "metastore.png", package = "brickster"),
contains = (
catalog = (
icon = ("icons", "catalog.png", package = "brickster"),
contains = (
schema = (
icon = ("icons", "schema.png", package = "brickster"),
contains = (
tables = (contains = (
= (
icon = ("icons", "table.png", package = "brickster"),
contains = (
metadata = (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
),
columns = (
contains = "data"
)
))
)),
volumes = (contains = (
volume = (
icon = ("icons", "volume.png", package = "brickster"),
contains = "data"
)
)),
models = (
contains = (
model = (
icon = ("icons", "model.png", package = "brickster"),
contains = (
metadata = (contains = "data"),
versions = (contains = (
= (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
)
))
))
)),
funcs = (contains = (
func = (
icon = ("icons", "func.png", package = "brickster"),
contains = "data"
)
))
))
))
)),
experiments = (
icon = ("icons", "exp.png", package = "brickster"),
contains = (
experiment = (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
)
)),
modelregistry = (
icon = ("icons", "model.png", package = "brickster"),
contains = (
model = (contains = (
metadata = (contains = "data"),
versions = (contains = (
= (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
)
))
))
)),
warehouses = (
icon = ("icons", "warehouse.png", package = "brickster"),
contains = (
warehouse = (
icon = ("icons", "open.png", package = "brickster"),
contains = "data"
)
))
))
)
}
# nocov end
