mergin in main branch

Merge remote-tracking branch 'origin/main'

NAMESPACE

@@ -5,11 +5,13 @@ S3method(execute,sewage_joiner)
 S3method(execute,sewage_node)
 S3method(execute,sewage_splitter)
 S3method(print,sewage_pipeline)
+S3method(pull_output,sewage_pipeline)
 export(Joiner)
 export(Pipeline)
 export(Splitter)
 export(add_node)
 export(draw)
 export(execute)
+export(pull_output)
 export(run)
 importFrom(glue,glue)

R/pipeline.R

@@ -19,7 +19,7 @@ is_pipeline = function(x) {
   inherits(x, 'sewage_pipeline')
 }
 
-' Run a pipeline
+#' Run a pipeline
 #'
 #' This function is the extry point for executing a pipeline object
 #' @param pipeline an initialized pipeline object
@@ -54,3 +54,33 @@ is_executed_pipeline = function(x) {
 
   return(length(x$outputs) > 0)
 }
+
+#' Extract output components from a pipeline
+#' @param x an executed pipeline object
+#' @param component a character string specifying which output component to pull
+#' #' \dontrun{
+#' pipeline = Pipeline()
+#' pipeline = pipeline |>
+#'     add_node(name = 'Splitter', component = Splitter(), input = 'file')
+#' result = run(pipeline, file = mtcars)
+#' result$outputs$Splitter.output_1
+#' result$outputs$Splitter.output_2
+#' }
+#' @export
+pull_output = function(x, component, ...) {
+  UseMethod("pull_output")
+}
+
+#' @export
+#' @rdname pull_output
+pull_output.sewage_pipeline = function(x, component, ...) {
+  if(!is_executed_pipeline(x)) {
+    stop("No outputs available. Please execute pipeline using 'run'")
+  }
+  if(!is.character(component)) {
+    stop("component must be a character string")
+  }
+  output = x$outputs[[component]]
+  print(output)
+}
+

README.Rmd

@@ -89,11 +89,11 @@ print(result)
 We can now access the results of our terminating nodes. A terminating node is any node that is not specified as input. By default when the pipeline is run, each node will overwrite the output of its input node. Therefore any node that is not fed forward to a new node will return output. In the case of this pipeline, the `Subsetter` and `Summarizer` edges are our terminating nodes. Therefore, we can access their results in the `outputs` object of the pipeline
 
 ```{r}
-result$outputs$Subsetter
+pull_output(result, "Subsetter")
 ```
 
 ```{r}
-result$outputs$Summarizer
+pull_output(result, "Summarizer")
 ```
 
 ## Multiple Entrypoints

README.md

@@ -127,7 +127,7 @@ edges are our terminating nodes. Therefore, we can access their results
 in the `outputs` object of the pipeline
 
 ``` r
-result$outputs$Subsetter
+pull_output(result, "Subsetter")
 #> # A tibble: 7 × 12
 #>   ...1           mpg   cyl  disp    hp  drat    wt  qsec    vs    am  gear  carb
 #>   <chr>        <dbl> <dbl> <dbl> <dbl> <dbl> <dbl> <dbl> <dbl> <dbl> <dbl> <dbl>
@@ -141,7 +141,7 @@ result$outputs$Subsetter
 ```
 
 ``` r
-result$outputs$Summarizer
+pull_output(result, "Summarizer")
 #>    Min. 1st Qu.  Median    Mean 3rd Qu.    Max. 
 #>    71.1   120.8   196.3   230.7   326.0   472.0
 ```

tests/testthat/test-pipeline.R

@@ -11,3 +11,21 @@ test_that("run fails for non-pipelines", {
     regexp = "pipeline object must be of type 'sewage_pipeline'"
     )
 })
+
+test_that("pull_output works for executed pipeline objects", {
+  input = "x"
+  pipeline = Pipeline() |>
+    add_node(component = c, name = "Printer", input = "x")
+  result = run(pipeline, x = input)
+  expect_identical(
+    pull_output(result, "Printer"),
+    c(input)
+  )
+})
+
+test_that("pull_output fails for non-executed pipelines", {
+  pipeline = Pipeline()
+  expect_error(
+    pull_output(pipeline, "Foo")
+  )
+})