added row policy fns

NAMESPACE

@@ -2,11 +2,16 @@
 
 S3method(as_priv,privilege)
 S3method(as_priv,tbl_sql)
+S3method(as_row_policy,row_policy)
+S3method(as_row_policy,tbl_sql)
 S3method(print,privilege)
 S3method(print,rls_policy)
+S3method(print,row_policy)
 export("%>%")
 export(as_priv)
+export(as_row_policy)
 export(auto_pipe)
+export(commands)
 export(from)
 export(grant)
 export(has_postgres)
@@ -27,6 +32,9 @@ export(rls_privileges)
 export(rls_run)
 export(rls_table_privileges)
 export(rls_tbl)
+export(row_policy)
+export(rows_existing)
+export(rows_new)
 export(to)
 export(translate_privilege)
 import(dbplyr)
@@ -38,6 +46,7 @@ importFrom(cli,cat_line)
 importFrom(cli,cli_abort)
 importFrom(cli,format_error)
 importFrom(dbplyr,sql)
+importFrom(dbplyr,translate_sql)
 importFrom(dplyr,"%>%")
 importFrom(dplyr,filter)
 importFrom(dplyr,tbl)
@@ -53,4 +62,6 @@ importFrom(rlang,enquos)
 importFrom(rlang,has_length)
 importFrom(rlang,is_character)
 importFrom(rlang,is_empty)
+importFrom(rlang,is_scalar_character)
+importFrom(rlang,quo_is_null)
 importFrom(tibble,as_tibble)

R/as_priv.R

@@ -36,12 +36,3 @@ cat_me <- function(x, y, indent = "  ") {
   y <- paste0(y, collapse = ", ")
   cat_line(glue("{indent}{x}: {y}", .trim = FALSE))
 }
-
-rls_grant <- function(commands, cols) {
-  x <- list(commands = commands, cols = cols)
-  structure(x, class = "rls_grant")
-}
-rls_revoke <- function(commands, cols) {
-  x <- list(commands = commands, cols = cols)
-  structure(x, class = "rls_revoke")
-}

R/as_row_policy.R

@@ -0,0 +1,46 @@
+#' As row policy
+#' @param x some input
+#' @export
+as_row_policy <- function(x) {
+  UseMethod("as_row_policy")
+}
+#' @export
+as_row_policy.row_policy <- function(x) {
+  return(x)
+}
+#' @export
+as_row_policy.tbl_sql <- function(x) {
+  tmp <- list(
+    data = x,
+    name = NULL,
+    commands = NULL,
+    user = NULL,
+    existing_rows = NULL,
+    new_rows = NULL,
+    type = NULL
+  )
+  structure(tmp, class = "row_policy")
+}
+#' @export
+print.row_policy <- function(x, ...) {
+  cat_line(glue("<row_policy> {x$name}"))
+  if (!is_really_empty(x$user)) {
+    cat_me("user", x$user)
+  }
+  if (!is_really_empty(x$commands)) {
+    cat_me("commands", x$commands)
+  }
+  if (!is_really_empty(x$existing_rows)) {
+    cat_me("existing rows", x$existing_rows)
+  }
+  if (!is_really_empty(x$new_rows)) {
+    cat_me("new rows", x$new_rows)
+  }
+  if (!is_really_empty(x$privilege)) {
+    cat_me("type", x$type)
+    for (i in x$privilege) {
+      cat_me(x = i$commands, y = i$cols %|||% "<all cols>", indent = "    ")
+    }
+  }
+  print(x$data)
+}

R/create.R

@@ -44,16 +44,16 @@ rls_create_policy <- function(con, policy) {
   )
   sql_create_policy <- glue("
     {create_statement} POLICY {policy$name} ON {policy$table}
-    {combine_if('FOR', policy$command)}
-    {combine_if('TO', policy$role)}
-    {combine_if('USING', policy$using)}
-    {combine_if('WITH CHECK', policy$check)}
+    {combine_if_old('FOR', policy$command)}
+    {combine_if_old('TO', policy$role)}
+    {combine_if_old('USING', policy$using)}
+    {combine_if_old('WITH CHECK', policy$check)}
   ")
   sql_create_policy <- gsub("\n\\s+\n", "\n", sql_create_policy)
   invisible(dbExecute(con, sql_create_policy))
 }
 
 # {ifelse(!is.null(policy$for_), paste('FOR', policy$for_), '')}
-combine_if <- function(statement, item) {
+combine_if_old <- function(statement, item) {
   ifelse(!is.null(item), paste(statement, item), "")
 }

R/pipeline.R

@@ -73,7 +73,7 @@ pipe_autoexec <- function(toggle) {
   info <- pipeline_info()
 
   if (isTRUE(info[["is_piped"]])) {
-    rls_exit <- function(x) if (inherits(x, "privilege")) rls_run(x@data$src$con, x) else x
+    rls_exit <- function(x) if (inherits(x, c("privilege", "row_policy"))) rls_run(x$data$src$con, x) else x
     pipeline_on_exit(info$env)
     info$env$.rls_exitfun <- if (toggle) rls_exit else identity
   }

R/privileges.R

@@ -72,7 +72,10 @@ revoke <- function(.data, ..., cols = NULL) {
 #' rls_tbl(con, "passwd") %>% to(jane, bob, alice)
 to <- function(.data, ...) {
   pipe_autoexec(toggle = rls_env$auto_pipe)
-  .data <- as_priv(.data)
+  .data <- switch(class(.data),
+    privilege = as_priv(.data),
+    row_policy = as_row_policy(.data)
+  )
   .data$user <- dot_names(...)
   .data
 }
@@ -161,9 +164,24 @@ priv_templates <- list(
 #' Run a query
 #'
 #' @export
-#' @param priv an s3 object of class `privilege`, required
+#' @param query an s3 object of class `privilege` or `row_policy, required
 #' @param con DBI connection object, required
-rls_run <- function(con, priv) {
-  sql <- translate_privilege(priv, con)
+rls_run <- function(con, query) {
+  is_conn(con)
+  assert_is(query, c("privilege", "row_policy"))
+  sql <- switch(class(query),
+    privilege = translate_privilege(query, con),
+    row_policy = translate_row_policy(query, con)
+  )
   dbExecute(con, sql)
 }
+
+rls_grant <- function(commands, cols) {
+  x <- list(commands = commands, cols = cols)
+  structure(x, class = "rls_grant")
+}
+
+rls_revoke <- function(commands, cols) {
+  x <- list(commands = commands, cols = cols)
+  structure(x, class = "rls_revoke")
+}

R/rls-package.R

@@ -7,6 +7,6 @@
 #' @importFrom RPostgres Postgres
 #' @importFrom tibble as_tibble
 #' @importFrom dplyr %>%
-#' @importFrom rlang as_name enquo enquos is_character
+#' @importFrom rlang as_name enquo enquos is_character quo_is_null
 ## usethis namespace: end
 NULL

R/row_policy.R

@@ -0,0 +1,132 @@
+#' Row policy
+#' 
+#' @export
+#' @inheritParams grant
+#' @param name (character) scalar name for the policy. required
+#' @examplesIf interactive() && has_postgres()
+#' library(RPostgres)
+#' con <- dbConnect(Postgres())
+#' rls_tbl(con, "passwd") %>%
+#'   row_policy("my_policy")
+row_policy <- function(.data, name) {
+  pipe_autoexec(toggle = rls_env$auto_pipe)
+  assert_is(name, "character")
+  assert_scalar(name)
+  .data <- as_row_policy(.data)
+  .data$name <- name
+  .data
+}
+
+#' Commands
+#' 
+#' @export
+#' @inheritParams grant
+#' @examplesIf interactive() && has_postgres()
+#' library(RPostgres)
+#' con <- dbConnect(Postgres())
+#' rls_tbl(con, "passwd") %>%
+#'   row_policy("my_policy") %>%
+#'   commands(update)
+commands <- function(.data, ...) {
+  pipe_autoexec(toggle = rls_env$auto_pipe)
+  .data <- as_row_policy(.data)
+  .data$commands <- dot_names(...)
+  .data
+}
+
+#' Create rule for existing rows
+#' 
+#' @export
+#' @inheritParams grant
+#' @param using an expression to use to check against existing rows
+#' @param sql (character) sql syntax to use for existing rows
+#' @details Use either `using` or `sql`, not both
+#' @examplesIf interactive() && has_postgres()
+#' library(RPostgres)
+#' con <- dbConnect(Postgres())
+#' rls_tbl(con, "passwd") %>%
+#'   row_policy("my_policy") %>%
+#'   commands(update) %>%
+#'   rows_existing(sql = 'current_user = "user_name"')
+rows_existing <- function(.data, using = NULL, sql = NULL) {
+  pipe_autoexec(toggle = rls_env$auto_pipe)
+  using_quo <- enquo(using)
+  stopifnot("Can not using and sql parameters together" =
+    xor(!rlang::quo_is_null(using_quo), !is_empty(sql)))
+  .data <- as_row_policy(.data)
+  if (rlang::is_null(sql)) {
+    .data$existing_rows <- translate_sql(!!using_quo, con = as_con(.data))
+  } else {
+    .data$existing_rows <- sql
+  }
+  .data
+}
+
+#' Create rule for new rows
+#' 
+#' @export
+#' @importFrom dbplyr translate_sql
+#' @inheritParams grant
+#' @param check an expression to use to check against addition of
+#' new rows or editing of existing rows
+#' @param sql (character) sql syntax to use for new rows
+#' @details Use either `check` or `sql`, not both
+#' @examplesIf interactive() && has_postgres()
+#' library(RPostgres)
+#' con <- dbConnect(Postgres())
+#'
+#' rls_tbl(con, "passwd") %>%
+#'   row_policy("a_policy") %>%
+#'   commands(update) %>%
+#'   rows_existing(TRUE) %>%
+#'   rows_new(TRUE) %>%
+#'   to(jane)
+#'
+#' rls_tbl(con, "passwd") %>%
+#'   row_policy("my_policy") %>%
+#'   commands(update) %>%
+#'   rows_existing(sql = 'current_user = "user_name"') %>%
+#'   rows_new(home_phone == "098-765-4321") %>%
+#'   to(jane)
+rows_new <- function(.data, check = NULL, sql = NULL) {
+  pipe_autoexec(toggle = rls_env$auto_pipe)
+  check_quo <- enquo(check)
+  stopifnot("Can not check and sql parameters together" =
+    xor(!rlang::quo_is_null(check_quo), !is_empty(sql)))
+  .data <- as_row_policy(.data)
+  if (rlang::is_null(sql)) {
+    .data$new_rows <- translate_sql(!!check_quo, con = as_con(.data))
+  } else {
+    .data$new_rows <- sql
+  }
+  .data
+}
+
+as_con <- function(x) {
+  assert_is(x, "row_policy")
+  x$data$src$con
+}
+
+combine_if <- function(statement, item, fun = \(x) x) {
+  ifelse(!rlang::is_null(item), paste(statement, fun(item)), "")
+}
+
+express <- function(x) {
+  glue("({ifelse(x == 'TRUE', tolower(x), x)})")
+}
+
+translate_row_policy <- function(policy, con) {
+  is_conn(con)
+  create_statement <- switch(class(con),
+    RedshiftConnection = "CREATE RLS",
+    PqConnection = "CREATE"
+  )
+  sql_create_policy <- glue("
+    {create_statement} POLICY {policy$name} ON {attr(policy$data, 'table')}
+    {combine_if('FOR', policy$commands)}
+    {combine_if('TO', policy$user)}
+    {combine_if('USING', policy$existing_rows, express)}
+    {combine_if('WITH CHECK', policy$new_rows, express)}
+  ")
+  sql(gsub("\n\\s+\n", "\n", sql_create_policy))
+}

R/utils.R

@@ -28,6 +28,15 @@ assert_is <- function(x, y, arg = caller_arg(x)) {
   }
 }
 
+#' @importFrom rlang is_scalar_character
+assert_scalar <- function(x, arg = caller_arg(x)) {
+  if (!is_scalar_character(x)) {
+    rls_abort(
+      format_error("{.arg {arg}} must be scalar")
+    )
+  }
+}
+
 #' @importFrom rlang has_length
 assert_len <- function(x, y, arg = caller_arg(x)) {
   if (!has_length(x, y)) {