V09 Read an API source

Use your current course session. Preparation and course runner · R script ↓

Optional deep dive. Enable only the options named in each block, after preparing its packages or services.

V09a Read a database source

Goal: Try a DBI backend locally.

Prerequisites: H04; adapters, RSQLite; extras.

The training database is in-memory SQLite. The owner of the connection must close it later.

training_step(
  "V09a",
  {
    adapter_con <- DBI::dbConnect(RSQLite::SQLite(), ":memory:")
    DBI::dbWriteTable(adapter_con, "policies", policies)
    database_source <- dataraft.adapters::dr_source_database(
      adapter_con,
      table = "policies",
      lazy = FALSE
    )
    database_rows <- dr_read_source(database_source)
    print(database_rows)
  },
  needs = c("H04"),
  packages = c("dataraft.adapters", "RSQLite"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: Six rows. SQLite may return logical values as 0/1; align types deliberately before applying a contract.

Task: May the adapter close this caller-owned connection?

V09b Write to a database target

Goal: Publish checked data to a table.

Prerequisites: V09a.

SQLite stores logical values as INTEGER. The recipe and output contract represent this explicitly. A database target is not a managed lake release.

training_step(
  "V09b",
  {
    database_contract <- dr_contract(
      "policies.sqlite",
      c(policy_id = "integer", premium = "numeric", cancelled = "integer"),
      key = "policy_id"
    )
    database_recipe <- dr_recipe() |>
      dr_step_mutate(cancelled = as.integer(cancelled))
    database_product <- dr_product(
      "policies.sqlite",
      policies,
      contract = database_contract
    ) |>
      dr_add_recipe(database_recipe) |>
      dr_add_quality(~ premium >= 0) |>
      dr_set_target(dataraft.adapters::dr_target_database(
        adapter_con,
        "checked_policies"
      ))
    database_result <- dr_publish(database_product)
    print(DBI::dbReadTable(adapter_con, "checked_policies"))
  },
  needs = c("V09a"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: An output table in the training database.

Task: Does this automatically provide DuckLake history?

V09c Transform with SQL

Goal: Prepare a table using DuckDB SQL.

Prerequisites: V09a.

The SQL adapter exposes the input through its documented table placeholder.

training_step(
  "V09c",
  {
    sql_source <- policies
    sql_recipe <- dr_recipe() |>
      dr_step_transform(dataraft.adapters::dr_sql_transform(
        "SELECT * FROM data WHERE premium >= 100"
      ))
    sql_product <- dr_product("policies.sql", sql_source) |>
      dr_add_recipe(sql_recipe)
    print(dr_collect(dr_run(sql_product, write = FALSE)))
  },
  needs = c("V09a"),
  packages = c("duckdb"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: Four policies with premiums of at least 100.

Task: Is R code automatically translated into SQL?

V09d Write Parquet

Goal: Change the output file format.

Prerequisites: H07; adapters, arrow; extras.

Parquet is a storage choice; the product and its rules remain reusable.

training_step(
  "V09d",
  {
    parquet_path <- file.path(training_root, "policies.parquet")
    parquet_product <- policy_product |>
      dr_set_target(dataraft.adapters::dr_target_parquet(
        parquet_path,
        overwrite = TRUE
      ))
    parquet_run <- dr_publish(parquet_product)
    print(parquet_run$outputs)
  },
  needs = c("H07"),
  packages = c("dataraft.adapters", "arrow"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: A Parquet output in the course directory.

Task: Why use overwrite=TRUE here?

V09e Read Parquet

Goal: Read stored columns back.

Prerequisites: V09d.

Reading is a separate source operation.

training_step(
  "V09e",
  {
    parquet_source <- dataraft.adapters::dr_source_parquet(parquet_path)
    parquet_rows <- dr_collect(dr_run(
      dr_product("policies.parquet", parquet_source),
      write = FALSE
    ))
    print(parquet_rows)
    stopifnot(nrow(parquet_rows) == 6L)
  },
  needs = c("V09d"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: Six rows.

Task: Is a filename a lake release ID?

V09f Publish to a pins board

Goal: Use a versioned board.

Prerequisites: H07; adapters, pins; extras.

A temporary board is enough for this local integration exercise.

training_step(
  "V09f",
  {
    training_board <- pins::board_temp(versioned = TRUE)
    pins_product <- policy_product |>
      dr_set_target(dataraft.adapters::dr_target_pins(
        training_board,
        "policies"
      ))
    pins_run <- dr_publish(pins_product)
    print(pins_run$outputs)
  },
  needs = c("H07"),
  packages = c("dataraft.adapters", "pins"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: A pin in the temporary board.

Task: Will this board persist after the session?

V09g Read a pin

Goal: Use a board as a source.

Prerequisites: V09f.

Without an explicit version, the source reads the current pin.

training_step(
  "V09g",
  {
    pins_source <- dataraft.adapters::dr_source_pins(training_board, "policies")
    pins_rows <- dr_collect(dr_run(
      dr_product("policies.pin", pins_source),
      write = FALSE
    ))
    print(pins_rows)
    stopifnot(nrow(pins_rows) == 6L)
  },
  needs = c("V09f"),
  enabled = training_options$extras,
  optional = TRUE
)

Expected result: The familiar policy delivery.

Task: How do you pin an earlier version?

V09 Read an API source

Goal: Execute an HTTP request deliberately.

Prerequisites: api=TRUE; adapters, httr2; TRAINING_API_URL returning a JSON list of records.

Use your own test API. No public example service is assumed. Add authentication to the request if needed, rather than storing it in the product.

training_step(
  "V09",
  {
    api_url <- Sys.getenv("TRAINING_API_URL")
    stopifnot(nzchar(api_url))
    api_request <- httr2::request(api_url) |> httr2::req_timeout(15)
    api_source <- dataraft.adapters::dr_source_api(
      api_request,
      parse = function(response) {
        dplyr::bind_rows(httr2::resp_body_json(
          response,
          simplifyVector = FALSE
        ))
      }
    )
    api_product <- dr_product("policies.api", api_source)
    print(dr_collect(dr_run(api_product, write = FALSE)))
  },
  packages = c("dataraft.adapters", "httr2"),
  enabled = training_options$api,
  optional = TRUE
)

Expected result: A table from your test response; no universal response schema is assumed.

Task: What if the response wraps records in a data field?