datafusion-server 0.21.0

Web server library for session-based queries using Arrow and other large datasets as data sources.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

// header.rs - Request header handler
// Sasaki, Naoki <nsasaki@sal.co.jp> February 21, 2024
//

use axum::http::header;
use axum_extra::{
    headers::{self, Header, HeaderName, HeaderValue},
    TypedHeader,
};

use crate::request::body::ResponseFormat;
use crate::response::http_error::ResponseError;

#[derive(Debug)]
pub struct Accept(String);

impl Header for Accept {
    fn name() -> &'static HeaderName {
        &header::ACCEPT
    }

    fn decode<'i, I>(values: &mut I) -> Result<Self, headers::Error>
    where
        I: Iterator<Item = &'i HeaderValue>,
    {
        values
            .next()
            .and_then(|value| value.to_str().ok())
            .map(|value| Accept(value.to_string()))
            .ok_or_else(headers::Error::invalid)
    }

    fn encode<E: Extend<HeaderValue>>(&self, values: &mut E) {
        let value = HeaderValue::from_str(&self.0).expect("Invalid accept header");
        values.extend(std::iter::once(value));
    }
}

pub fn response_format(
    accept_header: &TypedHeader<Accept>,
) -> Result<ResponseFormat, ResponseError> {
    let mut values: Vec<HeaderValue> = Vec::new();
    accept_header.0.encode(&mut values);

    let media_type = values
        .first()
        .ok_or_else(|| ResponseError::request_validation("Accept header should have a value"))?
        .to_str()
        .map_err(|_| ResponseError::internal_server_error("Failed to access the accept header"))?;

    match media_type {
        "application/json" => Ok(ResponseFormat::Json),
        "text/csv" => Ok(ResponseFormat::Csv),
        "application/vnd.apache.arrow.stream" => Ok(ResponseFormat::Arrow),
        _ => Err(ResponseError::unsupported_format(format!(
            "Unsupported response format '{media_type}'"
        ))),
    }
}

#[derive(Debug)]
pub struct ContentType(String);

impl Header for ContentType {
    fn name() -> &'static HeaderName {
        &header::CONTENT_TYPE
    }

    fn decode<'i, I>(values: &mut I) -> Result<Self, headers::Error>
    where
        I: Iterator<Item = &'i HeaderValue>,
    {
        values
            .next()
            .and_then(|value| value.to_str().ok())
            .map(|value| ContentType(value.to_string()))
            .ok_or_else(headers::Error::invalid)
    }

    fn encode<E: Extend<HeaderValue>>(&self, values: &mut E) {
        let value = HeaderValue::from_str(&self.0).expect("Invalid content-type header");
        values.extend(std::iter::once(value));
    }
}

pub fn request_format(content_type: &TypedHeader<ContentType>) -> Result<String, ResponseError> {
    let mut values: Vec<HeaderValue> = Vec::new();
    content_type.0.encode(&mut values);

    Ok(values
        .first()
        .ok_or_else(|| {
            ResponseError::request_validation("content-type header should have a value")
        })?
        .to_str()
        .map_err(|_| {
            ResponseError::internal_server_error("Failed to access the content-type header")
        })?
        .to_string())
}