Examples

These examples assume the listed ports have already been declared on the processing block.

Apply a JSON threshold to a table

Ports:

Direction	Name	Type
Input	rows	Arrow table
Input	config	JSON
Output	accepted	Arrow table
Output	summary	JSON

import pyarrow.compute as pc

import flasktrack as ft

rows = ft.input_table("rows")
config = ft.input_json("config")

if "value" not in rows.column_names:
    raise ValueError("rows is missing the value column")
if not isinstance(config, dict) or "minimum_value" not in config:
    raise ValueError("config must contain minimum_value")

minimum = float(config["minimum_value"])
mask = pc.and_(pc.is_valid(rows["value"]), pc.greater_equal(rows["value"], minimum))
accepted = rows.filter(mask)

ft.output_table("accepted", accepted)
ft.output_json(
    "summary",
    {
        "minimum_value": minimum,
        "input_rows": rows.num_rows,
        "accepted_rows": accepted.num_rows,
    },
)

Produce a CSV export

Ports:

Direction	Name	Type
Input	results	Arrow table
Output	csv_export	File

import flasktrack as ft

results = ft.input_table("results")
ft.output_csv("csv_export", results)

R summary with dplyr

Ports:

Direction	Name	Type
Input	observations	Arrow table
Output	summary	Arrow table

source("/opt/flasktrack/flasktrack.R")

observations <- ft_input_table("observations")
required <- c("sample_id", "value")
missing <- setdiff(required, names(observations))
if (length(missing) > 0L) {
  stop(paste("Missing required columns:", paste(missing, collapse = ", ")))
}

summary <- observations |>
  dplyr::filter(!is.na(.data$value)) |>
  dplyr::group_by(.data$sample_id) |>
  dplyr::summarise(
    value_mean = mean(.data$value),
    value_sd = stats::sd(.data$value),
    observation_count = dplyr::n(),
    .groups = "drop"
  )

ft_output_table("summary", summary)
ft_log(sprintf("Published %d summary rows", nrow(summary)))

Transform a text file

Ports:

Direction	Name	Type
Input	source_text	File
Output	normalized_text	File

import flasktrack as ft

text = ft.input_text("source_text")
normalized = "\n".join(
    line.strip() for line in text.splitlines() if line.strip()
) + "\n"
ft.output_text("normalized_text", normalized)

Publish a generated report

When another library creates a PDF, image, workbook, or specialized artifact, publish the completed path through the declared file output:

import flasktrack as ft

report_path = build_report()
ft.output_file("report", report_path, "application/pdf")

The build_report() function represents your own report-generation code.