Building Data CLIs & Services

use anyhow::{Context, Result};
use clap::Parser;
use polars::prelude::*;
use tracing::info;
 
#[derive(Parser)]
#[command(name = "csv2pq", about = "Convert CSV to Parquet with optional filter")]
struct Cli {
    #[arg(long)]
    input: String,
    #[arg(long)]
    output: String,
    #[arg(long, default_value_t = 0)]
    min_amount: i64,
}
 
fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    let cli = Cli::parse();
    run(&cli).context("etl failed")?;
    Ok(())
}
 
fn run(cli: &Cli) -> PolarsResult<()> {
    info!(input = %cli.input, "starting");
    let mut df = CsvReadOptions::default()
        .try_into_reader_with_file_path(Some(cli.input.clone().into()))?
        .finish()?;
    if cli.min_amount > 0 {
        df = df.lazy().filter(col("amount").gt(lit(cli.min_amount))).collect()?;
    }
    ParquetWriter::new(std::fs::File::create(&cli.output)?).finish(&mut df)?;
    info!(rows = df.height(), "done");
    Ok(())
}

use axum::{extract::Query, routing::get, Json, Router};
use polars::prelude::*;
use serde::Deserialize;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::RwLock;
 
#[derive(Clone)]
struct AppState {
    df: Arc<RwLock<DataFrame>>,
}
 
#[derive(Deserialize)]
struct Params {
    region: String,
}
 
async fn summary(
    Query(p): Query<Params>,
    axum::extract::State(state): axum::extract::State<AppState>,
) -> Result<Json<serde_json::Value>, String> {
    let df = state.df.read().await;
    let filtered = df
        .clone()
        .lazy()
        .filter(col("region").eq(lit(p.region.clone())))
        .select([col("amount").sum().alias("total")])
        .collect()
        .map_err(|e| e.to_string())?;
    let total: i64 = filtered.column("total").unwrap().get(0).unwrap().try_extract().unwrap();
    Ok(Json(serde_json::json!({ "region": p.region, "total": total })))
}
 
#[tokio::main]
async fn main() -> PolarsResult<()> {
    let df = CsvReadOptions::default()
        .try_into_reader_with_file_path(Some("sales.csv".into()))?
        .finish()?;
    let state = AppState { df: Arc::new(RwLock::new(df)) };
    let app = Router::new().route("/summary", get(summary)).with_state(state);
    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
    Ok(())
}

// Distinguish exit codes in CLI main:
// std::process::exit(2) for bad args, 1 for data errors - helps orchestrators retry correctly.