Repository: roapi/roapi (language: python)

Issue:
Changing the default response serialization in the config.yaml
For certain use-cases [setting an http header](https://roapi.github.io/docs/api/response.html) is awkward or impossible, e.g. when specifying the data source in a vega or vega-lite visualization ([you can set what format to parse as, but not the header](https://vega.github.io/vega-lite/docs/data.html#url)). Therefore it would be nice if it were possible to change the default response serialization in the config.yaml.

Changing the default format from `application/json` to `application/vnd.apache.arrow.stream` could look something like this:
```yaml
addr:
   http: 0.0.0.0:8084
response_format: "vnd.apache.arrow.stream"
tables:
  - name: "blogs"
    uri: "test_data/blogs.parquet"
```




Discussion / hints:
Good idea, I think the change should be pretty straightforward. We can add a config and pass it all the way to: https://github.com/roapi/roapi/blob/193ea79878ad1eda48210356cc0012b554cbce10/roapi/src/api/mod.rs#L30

As a quick workaround, we can also put a reverse proxy in front of roapi to set the content-type header by default.

Fix patch:
diff --git a/columnq/src/encoding/mod.rs b/columnq/src/encoding/mod.rs
index ccabff731..e5cdbc916 100644
--- a/columnq/src/encoding/mod.rs
+++ b/columnq/src/encoding/mod.rs
@@ -1,6 +1,9 @@
+use serde_derive::Deserialize;
 use std::convert::TryFrom;
 
+#[derive(Deserialize, Default, Clone, Copy)]
 pub enum ContentType {
+    #[default]
     Json,
     Csv,
     ArrowFile,
diff --git a/roapi/src/api/graphql.rs b/roapi/src/api/graphql.rs
index 57c491edf..ab9aba481 100644
--- a/roapi/src/api/graphql.rs
+++ b/roapi/src/api/graphql.rs
@@ -14,7 +14,8 @@ pub async fn post<H: RoapiContext>(
     body: Bytes,
 ) -> Result<impl IntoResponse, ApiErrResp> {
     let ctx = state.0;
-    let encode_type = encode_type_from_hdr(headers);
+    let response_format = H::get_response_format(&ctx).await;
+    let encode_type = encode_type_from_hdr(headers, response_format);
     let graphq = std::str::from_utf8(&body).map_err(ApiErrResp::read_query)?;
     let batches = ctx.query_graphql(graphq).await?;
     encode_record_batches(encode_type, &batches)
diff --git a/roapi/src/api/mod.rs b/roapi/src/api/mod.rs
index e672a39ea..82a1f5d78 100644
--- a/roapi/src/api/mod.rs
+++ b/roapi/src/api/mod.rs
@@ -6,7 +6,6 @@ use axum::http::Response;
 use axum::response::IntoResponse;
 use columnq::datafusion::arrow;
 use columnq::encoding;
-use columnq::encoding::ContentType;
 
 use crate::error::ApiErrResp;
 
@@ -25,11 +24,14 @@ pub fn bytes_to_json_resp(bytes: Vec<u8>) -> impl IntoResponse {
     bytes_to_resp(bytes, "application/json")
 }
 
-pub fn encode_type_from_hdr(headers: header::HeaderMap) -> encoding::ContentType {
+pub fn encode_type_from_hdr(
+    headers: header::HeaderMap,
+    response_format: encoding::ContentType,
+) -> encoding::ContentType {
     match headers.get(header::ACCEPT) {
-        None => encoding::ContentType::Json,
+        None => response_format,
         Some(hdr_value) => {
-            encoding::ContentType::try_from(hdr_value.as_bytes()).unwrap_or(ContentType::Json)
+            encoding::ContentType::try_from(hdr_value.as_bytes()).unwrap_or(response_format)
         }
     }
 }
diff --git a/roapi/src/api/rest.rs b/roapi/src/api/rest.rs
index 7a8ec1493..5370d25de 100644
--- a/roapi/src/api/rest.rs
+++ b/roapi/src/api/rest.rs
@@ -14,7 +14,8 @@ pub async fn get_table<H: RoapiContext>(
     extract::Path(table_name): extract::Path<String>,
     extract::Query(params): extract::Query<HashMap<String, String>>,
 ) -> Result<impl IntoResponse, ApiErrResp> {
-    let encode_type = encode_type_from_hdr(headers);
+    let response_format = H::get_response_format(&ctx).await;
+    let encode_type = encode_type_from_hdr(headers, response_format);
     let batches = ctx.query_rest_table(&table_name, &params).await?;
     encode_record_batches(encode_type, &batches)
 }
diff --git a/roapi/src/api/sql.rs b/roapi/src/api/sql.rs
index cfbde2daa..6afccfb28 100644
--- a/roapi/src/api/sql.rs
+++ b/roapi/src/api/sql.rs
@@ -15,7 +15,8 @@ pub async fn post<H: RoapiContext>(
     body: Bytes,
 ) -> Result<impl IntoResponse, ApiErrResp> {
     let ctx = state.0;
-    let encode_type = encode_type_from_hdr(headers);
+    let response_format = H::get_response_format(&ctx).await;
+    let encode_type = encode_type_from_hdr(headers, response_format);
     let sql = std::str::from_utf8(&body).map_err(ApiErrResp::read_query)?;
     let batches = ctx.query_sql(sql).await?;
     encode_record_batches(encode_type, &batches)
diff --git a/roapi/src/config.rs b/roapi/src/config.rs
index 487ba6f2b..3985fd049 100644
--- a/roapi/src/config.rs
+++ b/roapi/src/config.rs
@@ -2,6 +2,7 @@ use serde_derive::Deserialize;
 
 use anyhow::{bail, Context, Result};
 
+use columnq::encoding;
 use columnq::table::parse_table_uri_arg;
 use columnq::table::KeyValueSource;
 use columnq::table::TableSource;
@@ -23,6 +24,8 @@ pub struct Config {
     pub disable_read_only: bool,
     #[serde(default)]
     pub kvstores: Vec<KeyValueSource>,
+    #[serde(default)]
+    pub response_format: encoding::ContentType,
 }
 
 fn table_arg() -> clap::Arg<'static> {
@@ -75,6 +78,16 @@ fn reload_interval_arg() -> clap::Arg<'static> {
         .short('r')
 }
 
+fn response_format_arg() -> clap::Arg<'static> {
+    clap::Arg::new("response-format")
+        .help("change response serialization: Json (default), Csv, ArrowFile, ArrowStream, Parquet")
+        .required(false)
+        .takes_value(true)
+        .value_name("ResponseFormat")
+        .long("response-format")
+        .short('f')
+}
+
 fn config_arg() -> clap::Arg<'static> {
     clap::Arg::new("config")
         .help("config file path")
@@ -98,6 +111,7 @@ pub fn get_configuration() -> Result<Config, anyhow::Error> {
             config_arg(),
             read_only_arg(),
             reload_interval_arg(),
+            response_format_arg(),
             table_arg(),
         ])
         .get_matches();
@@ -139,5 +153,9 @@ pub fn get_configuration() -> Result<Config, anyhow::Error> {
         ));
     }
 
+    if let Some(response_format) = matches.value_of("response-format") {
+        config.response_format = serde_yaml::from_str(response_format)?;
+    }
+
     Ok(config)
 }
diff --git a/roapi/src/context.rs b/roapi/src/context.rs
index ac1eb012c..06e71dc2b 100644
--- a/roapi/src/context.rs
+++ b/roapi/src/context.rs
@@ -5,6 +5,7 @@ use async_trait::async_trait;
 use columnq::datafusion::arrow;
 use columnq::datafusion::dataframe::DataFrame;
 use columnq::datafusion::error::DataFusionError;
+use columnq::encoding;
 use columnq::error::ColumnQError;
 use columnq::error::QueryError;
 use columnq::table::TableSource;
@@ -17,6 +18,7 @@ use crate::error::ApiErrResp;
 
 pub struct RawRoapiContext {
     pub cq: ColumnQ,
+    pub response_format: encoding::ContentType,
     // TODO: store pre serialized schema in handler context
 }
 
@@ -40,7 +42,10 @@ impl RawRoapiContext {
             info!("registered `{}` as kv store `{}`", k.io_source, k.name);
         }
 
-        Ok(Self { cq })
+        Ok(Self {
+            cq,
+            response_format: config.response_format,
+        })
     }
 }
 
@@ -75,6 +80,8 @@ pub trait RoapiContext: Send + Sync + 'static {
     async fn kv_get(&self, kv_name: &str, key: &str) -> Result<Option<String>, QueryError>;
 
     async fn sql_to_df(&self, query: &str) -> Result<Arc<DataFrame>, DataFusionError>;
+
+    async fn get_response_format(&self) -> encoding::ContentType;
 }
 
 #[async_trait]
@@ -145,6 +152,11 @@ impl RoapiContext for RawRoapiContext {
     async fn sql_to_df(&self, query: &str) -> Result<Arc<DataFrame>, DataFusionError> {
         self.cq.dfctx.sql(query).await
     }
+
+    #[inline]
+    async fn get_response_format(&self) -> encoding::ContentType {
+        self.response_format
+    }
 }
 
 #[async_trait]
@@ -221,4 +233,10 @@ impl RoapiContext for ConcurrentRoapiContext {
         let ctx = self.read().await;
         ctx.cq.dfctx.sql(query).await
     }
+
+    #[inline]
+    async fn get_response_format(&self) -> encoding::ContentType {
+        let ctx = self.read().await;
+        ctx.response_format
+    }
 }


Test patch:
diff --git a/roapi/tests/helpers.rs b/roapi/tests/helpers.rs
index c98adec80..fde7db7c6 100644
--- a/roapi/tests/helpers.rs
+++ b/roapi/tests/helpers.rs
@@ -33,8 +33,8 @@ pub async fn test_api_app(
         },
         tables,
         reload_interval: Some(Duration::from_secs(1000)),
-        disable_read_only: false,
         kvstores,
+        ..Default::default()
     };
 
     let app = Application::build(config)