6299724219

Committed 25 Sep 2023 12:58PM UTC coverage: 77.81% (+0.5%) from 77.275%

Build # 6299724219

Build Type

push

github

Committed by

chubei

Commit Message

fix: Add `BINDGEN_EXTRA_CLANG_ARGS` to cross compile rocksdb

Run Details

50223 of 64546 relevant lines covered (77.81%)

148909.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.6

/dozer-api/src/rest/api_generator.rs

use std::sync::Arc;

use actix_web::web::ReqData;
use actix_web::{web, HttpResponse};
use dozer_cache::cache::expression::{QueryExpression, Skip};
use dozer_cache::cache::CacheRecord;
use dozer_cache::{CacheReader, Phase};
use dozer_types::errors::types::CannotConvertF64ToJson;
use dozer_types::indexmap::IndexMap;
use dozer_types::models::api_endpoint::ApiEndpoint;
use dozer_types::types::{Field, Schema};
use openapiv3::OpenAPI;

use crate::api_helper::{get_record, get_records, get_records_count};
use crate::generator::oapi::generator::OpenApiGenerator;
use crate::CacheEndpoint;
use crate::{auth::Access, errors::ApiError};
use dozer_types::grpc_types::health::health_check_response::ServingStatus;
use dozer_types::json_types::field_to_json_value;
use dozer_types::serde_json::{json, Value};

use self::extractor::QueryExpressionExtractor;

fn generate_oapi3(reader: &CacheReader, endpoint: ApiEndpoint) -> Result<OpenAPI, ApiError> {
    let (schema, secondary_indexes) = reader.get_schema();

    let oapi_generator = OpenApiGenerator::new(
        schema,
        secondary_indexes,
        endpoint,
        vec![format!("http://localhost:{}", "8080")],
    );

    Ok(oapi_generator.generate_oas3())
}

/// Generated function to return openapi.yaml documentation.
pub async fn generate_oapi(
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
) -> Result<HttpResponse, ApiError> {
    generate_oapi3(
        &cache_endpoint.cache_reader(),
        cache_endpoint.endpoint.clone(),
    )
    .map(|result| HttpResponse::Ok().json(result))
}

// Generated Get function to return a single record in JSON format
pub async fn get(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
    path: web::Path<String>,
) -> Result<HttpResponse, ApiError> {
    let cache_reader = &cache_endpoint.cache_reader();
    let schema = &cache_reader.get_schema().0;

    let key = path.as_str();
    let key = if schema.primary_index.is_empty() {
        return Err(ApiError::NoPrimaryKey);
    } else if schema.primary_index.len() == 1 {
        let field = &schema.fields[schema.primary_index[0]];
        Field::from_str(key, field.typ, field.nullable).map_err(ApiError::InvalidPrimaryKey)?
    } else {
        return Err(ApiError::MultiIndexFetch(key.to_string()));
    };

    // This implementation must be consistent with `dozer_cache::cache::index::get_primary_key`
    let key = key.encode();
    let record = get_record(
        &cache_endpoint.cache_reader(),
        &key,
        &cache_endpoint.endpoint.name,
        access.map(|a| a.into_inner()),
    )?;

    record_to_map(record, schema)
        .map(|map| HttpResponse::Ok().json(map))
        .map_err(Into::into)
}

// Generated list function for multiple records with a default query expression
pub async fn list(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
) -> Result<HttpResponse, ApiError> {
    let mut exp = QueryExpression::new(None, vec![], Some(50), Skip::Skip(0));
    get_records_map(access, cache_endpoint, &mut exp).map(|map| HttpResponse::Ok().json(map))
}

// Generated get function for health check
pub async fn health_route() -> Result<HttpResponse, ApiError> {
    let status = ServingStatus::Serving;
    let resp = json!({ "status": status.as_str_name() }).to_string();
    Ok(HttpResponse::Ok().body(resp))
}

pub async fn count(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
    query_expression: QueryExpressionExtractor,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = query_expression.0;
    get_records_count(
        &cache_endpoint.cache_reader(),
        &mut query_expression,
        &cache_endpoint.endpoint.name,
        access.map(|a| a.into_inner()),
    )
    .map(|count| HttpResponse::Ok().json(count))
}

// Generated query function for multiple records
pub async fn query(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
    query_expression: QueryExpressionExtractor,
    default_max_num_records: web::Data<usize>,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = query_expression.0;
    if query_expression.limit.is_none() {
        query_expression.limit = Some(**default_max_num_records);
    }

    get_records_map(access, cache_endpoint, &mut query_expression)
        .map(|maps| HttpResponse::Ok().json(maps))
}

/// Get multiple records
fn get_records_map(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
    exp: &mut QueryExpression,
) -> Result<Vec<IndexMap<String, Value>>, ApiError> {
    let mut maps = vec![];
    let cache_reader = &cache_endpoint.cache_reader();
    let records = get_records(
        cache_reader,
        exp,
        &cache_endpoint.endpoint.name,
        access.map(|a| a.into_inner()),
    )?;
    let schema = &cache_reader.get_schema().0;
    for record in records.into_iter() {
        let map = record_to_map(record, schema)?;
        maps.push(map);
    }
    Ok(maps)
}

/// Used in REST APIs for converting to JSON
fn record_to_map(
    record: CacheRecord,
    schema: &Schema,
) -> Result<IndexMap<String, Value>, CannotConvertF64ToJson> {
    let mut map = IndexMap::new();

    for (field_def, field) in schema.fields.iter().zip(record.record.values) {
        let val = field_to_json_value(field)?;
        map.insert(field_def.name.clone(), val);
    }

    map.insert("__dozer_record_id".to_string(), Value::from(record.id));
    map.insert(
        "__dozer_record_version".to_string(),
        Value::from(record.version),
    );

    Ok(map)
}

pub async fn get_phase(
    cache_endpoint: ReqData<Arc<CacheEndpoint>>,
) -> Result<web::Json<Phase>, ApiError> {
    let cache_reader = cache_endpoint.cache_reader();
    let phase = cache_reader.get_phase().map_err(ApiError::GetPhaseFailed)?;
    Ok(web::Json(phase))
}

mod extractor {
    use std::{
        future::{ready, Ready},
        task::Poll,
    };

    use actix_http::{HttpMessage, Payload};
    use actix_web::{
        error::{ErrorBadRequest, JsonPayloadError},
        Error, FromRequest, HttpRequest,
    };
    use dozer_cache::cache::expression::QueryExpression;
    use dozer_types::serde_json;
    use futures_util::{future::Either, Future};
    use pin_project::pin_project;

    pub struct QueryExpressionExtractor(pub QueryExpression);

    impl FromRequest for QueryExpressionExtractor {
        type Error = Error;
        type Future =
            Either<Ready<Result<QueryExpressionExtractor, Error>>, QueryExpressionExtractFuture>;

        fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
            if let Err(e) = check_content_type(req) {
                Either::Left(ready(Err(e)))
            } else {
                Either::Right(QueryExpressionExtractFuture(String::from_request(
                    req, payload,
                )))
            }
        }
    }

    fn check_content_type(req: &HttpRequest) -> Result<(), Error> {
        if let Some(mime_type) = req.mime_type()? {
            if mime_type != "application/json" {
                return Err(ErrorBadRequest("Content-Type is not application/json"));
            }
        }
        Ok(())
    }
    type StringExtractFut = <String as FromRequest>::Future;

    #[pin_project]
    pub struct QueryExpressionExtractFuture(#[pin] StringExtractFut);

    impl Future for QueryExpressionExtractFuture {
        type Output = Result<QueryExpressionExtractor, Error>;

        fn poll(
            self: std::pin::Pin<&mut Self>,
            cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Self::Output> {
            let this = self.project();
            match this.0.poll(cx) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
                Poll::Ready(Ok(body)) => {
                    Poll::Ready(parse_query_expression(&body).map(QueryExpressionExtractor))
                }
            }
        }
    }

    fn parse_query_expression(body: &str) -> Result<QueryExpression, Error> {
        if body.is_empty() {
            return Ok(QueryExpression::with_no_limit());
        }

        serde_json::from_str(body)
            .map_err(JsonPayloadError::Deserialize)
            .map_err(Into::into)
    }
}

1	use std::sync::Arc;
2
3	use actix_web::web::ReqData;
4	use actix_web::{web, HttpResponse};
5	use dozer_cache::cache::expression::{QueryExpression, Skip};
6	use dozer_cache::cache::CacheRecord;
7	use dozer_cache::{CacheReader, Phase};
8	use dozer_types::errors::types::CannotConvertF64ToJson;
9	use dozer_types::indexmap::IndexMap;
10	use dozer_types::models::api_endpoint::ApiEndpoint;
11	use dozer_types::types::{Field, Schema};
12	use openapiv3::OpenAPI;
13
14	use crate::api_helper::{get_record, get_records, get_records_count};
15	use crate::generator::oapi::generator::OpenApiGenerator;
16	use crate::CacheEndpoint;
17	use crate::{auth::Access, errors::ApiError};
18	use dozer_types::grpc_types::health::health_check_response::ServingStatus;
19	use dozer_types::json_types::field_to_json_value;
20	use dozer_types::serde_json::{json, Value};
21
22	use self::extractor::QueryExpressionExtractor;
23
24	fn generate_oapi3(reader: &CacheReader, endpoint: ApiEndpoint) -> Result<OpenAPI, ApiError> {	×
25	let (schema, secondary_indexes) = reader.get_schema();	×
26		×
27	let oapi_generator = OpenApiGenerator::new(	×
28	schema,	×
29	secondary_indexes,	×
30	endpoint,	×
31	vec![format!("http://localhost:{}", "8080")],	×
32	);	×
33		×
34	Ok(oapi_generator.generate_oas3())	×
35	}	×
36
37	/// Generated function to return openapi.yaml documentation.
38	pub async fn generate_oapi(	×
39	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	×
40	) -> Result<HttpResponse, ApiError> {	×
41	generate_oapi3(	×
42	&cache_endpoint.cache_reader(),	×
43	cache_endpoint.endpoint.clone(),	×
44	)	×
45	.map(\|result\| HttpResponse::Ok().json(result))	×
46	}	×
47
48	// Generated Get function to return a single record in JSON format
49	pub async fn get(	2✔
50	access: Option<ReqData<Access>>,	2✔
51	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	2✔
52	path: web::Path<String>,	2✔
53	) -> Result<HttpResponse, ApiError> {	2✔
54	let cache_reader = &cache_endpoint.cache_reader();	2✔
55	let schema = &cache_reader.get_schema().0;	2✔
56		2✔
57	let key = path.as_str();	2✔
58	let key = if schema.primary_index.is_empty() {	2✔
59	return Err(ApiError::NoPrimaryKey);	×
60	} else if schema.primary_index.len() == 1 {	2✔
61	let field = &schema.fields[schema.primary_index[0]];	2✔
62	Field::from_str(key, field.typ, field.nullable).map_err(ApiError::InvalidPrimaryKey)?	2✔
63	} else {
64	return Err(ApiError::MultiIndexFetch(key.to_string()));	×
65	};
66
67	// This implementation must be consistent with `dozer_cache::cache::index::get_primary_key`
68	let key = key.encode();	1✔
69	let record = get_record(	1✔
70	&cache_endpoint.cache_reader(),	1✔
71	&key,	1✔
72	&cache_endpoint.endpoint.name,	1✔
73	access.map(\|a\| a.into_inner()),	1✔
74	)?;	1✔
75
76	record_to_map(record, schema)	1✔
77	.map(\|map\| HttpResponse::Ok().json(map))	1✔
78	.map_err(Into::into)	1✔
79	}	2✔
80
81	// Generated list function for multiple records with a default query expression
82	pub async fn list(	3✔
83	access: Option<ReqData<Access>>,	3✔
84	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	3✔
85	) -> Result<HttpResponse, ApiError> {	3✔
86	let mut exp = QueryExpression::new(None, vec![], Some(50), Skip::Skip(0));	3✔
87	get_records_map(access, cache_endpoint, &mut exp).map(\|map\| HttpResponse::Ok().json(map))	3✔
88	}	3✔
89
90	// Generated get function for health check
91	pub async fn health_route() -> Result<HttpResponse, ApiError> {	×
92	let status = ServingStatus::Serving;	×
93	let resp = json!({ "status": status.as_str_name() }).to_string();	×
94	Ok(HttpResponse::Ok().body(resp))	×
95	}	×
96
97	pub async fn count(	7✔
98	access: Option<ReqData<Access>>,	7✔
99	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	7✔
100	query_expression: QueryExpressionExtractor,	7✔
101	) -> Result<HttpResponse, ApiError> {	7✔
102	let mut query_expression = query_expression.0;	7✔
103	get_records_count(	7✔
104	&cache_endpoint.cache_reader(),	7✔
105	&mut query_expression,	7✔
106	&cache_endpoint.endpoint.name,	7✔
107	access.map(\|a\| a.into_inner()),	7✔
108	)	7✔
109	.map(\|count\| HttpResponse::Ok().json(count))	7✔
110	}	7✔
111
112	// Generated query function for multiple records
113	pub async fn query(	7✔
114	access: Option<ReqData<Access>>,	7✔
115	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	7✔
116	query_expression: QueryExpressionExtractor,	7✔
117	default_max_num_records: web::Data<usize>,	7✔
118	) -> Result<HttpResponse, ApiError> {	7✔
119	let mut query_expression = query_expression.0;	7✔
120	if query_expression.limit.is_none() {	7✔
121	query_expression.limit = Some(**default_max_num_records);	6✔
122	}	6✔
123
124	get_records_map(access, cache_endpoint, &mut query_expression)	7✔
125	.map(\|maps\| HttpResponse::Ok().json(maps))	7✔
126	}	7✔
127
128	/// Get multiple records
129	fn get_records_map(	10✔
130	access: Option<ReqData<Access>>,	10✔
131	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	10✔
132	exp: &mut QueryExpression,	10✔
133	) -> Result<Vec<IndexMap<String, Value>>, ApiError> {	10✔
134	let mut maps = vec![];	10✔
135	let cache_reader = &cache_endpoint.cache_reader();	10✔
136	let records = get_records(	10✔
137	cache_reader,	10✔
138	exp,	10✔
139	&cache_endpoint.endpoint.name,	10✔
140	access.map(\|a\| a.into_inner()),	10✔
141	)?;	10✔
142	let schema = &cache_reader.get_schema().0;	10✔
143	for record in records.into_iter() {	412✔
144	let map = record_to_map(record, schema)?;	412✔
145	maps.push(map);	412✔
146	}
147	Ok(maps)	10✔
148	}	10✔
149
150	/// Used in REST APIs for converting to JSON
151	fn record_to_map(	413✔
152	record: CacheRecord,	413✔
153	schema: &Schema,	413✔
154	) -> Result<IndexMap<String, Value>, CannotConvertF64ToJson> {	413✔
155	let mut map = IndexMap::new();	413✔
156
157	for (field_def, field) in schema.fields.iter().zip(record.record.values) {	2,065✔
158	let val = field_to_json_value(field)?;	2,065✔
159	map.insert(field_def.name.clone(), val);	2,065✔
160	}
161
162	map.insert("__dozer_record_id".to_string(), Value::from(record.id));	413✔
163	map.insert(	413✔
164	"__dozer_record_version".to_string(),	413✔
165	Value::from(record.version),	413✔
166	);	413✔
167		413✔
168	Ok(map)	413✔
169	}	413✔
170
171	pub async fn get_phase(	1✔
172	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	1✔
173	) -> Result<web::Json<Phase>, ApiError> {	1✔
174	let cache_reader = cache_endpoint.cache_reader();	1✔
175	let phase = cache_reader.get_phase().map_err(ApiError::GetPhaseFailed)?;	1✔
176	Ok(web::Json(phase))	1✔
177	}	1✔
178
179	mod extractor {
180	use std::{
181	future::{ready, Ready},
182	task::Poll,
183	};
184
185	use actix_http::{HttpMessage, Payload};
186	use actix_web::{
187	error::{ErrorBadRequest, JsonPayloadError},
188	Error, FromRequest, HttpRequest,
189	};
190	use dozer_cache::cache::expression::QueryExpression;
191	use dozer_types::serde_json;
192	use futures_util::{future::Either, Future};
193	use pin_project::pin_project;
194
195	pub struct QueryExpressionExtractor(pub QueryExpression);
196
197	impl FromRequest for QueryExpressionExtractor {
198	type Error = Error;
199	type Future =
200	Either<Ready<Result<QueryExpressionExtractor, Error>>, QueryExpressionExtractFuture>;
201
202	fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
203	if let Err(e) = check_content_type(req) {	16✔
204	Either::Left(ready(Err(e)))	1✔
205	} else {
206	Either::Right(QueryExpressionExtractFuture(String::from_request(	15✔
207	req, payload,	15✔
208	)))	15✔
209	}
210	}	16✔
211	}
212
213	fn check_content_type(req: &HttpRequest) -> Result<(), Error> {	16✔
214	if let Some(mime_type) = req.mime_type()? {	16✔
215	if mime_type != "application/json" {	12✔
216	return Err(ErrorBadRequest("Content-Type is not application/json"));	1✔
217	}	11✔
218	}	4✔
219	Ok(())	15✔
220	}	16✔
221	type StringExtractFut = <String as FromRequest>::Future;
222
223	#[pin_project]	15✔
224	pub struct QueryExpressionExtractFuture(#[pin] StringExtractFut);
225
226	impl Future for QueryExpressionExtractFuture {
227	type Output = Result<QueryExpressionExtractor, Error>;
228
229	fn poll(	15✔
230	self: std::pin::Pin<&mut Self>,	15✔
231	cx: &mut std::task::Context<'_>,	15✔
232	) -> std::task::Poll<Self::Output> {	15✔
233	let this = self.project();	15✔
234	match this.0.poll(cx) {	15✔
235	Poll::Pending => Poll::Pending,	×
236	Poll::Ready(Err(e)) => Poll::Ready(Err(e)),	×
237	Poll::Ready(Ok(body)) => {	15✔
238	Poll::Ready(parse_query_expression(&body).map(QueryExpressionExtractor))	15✔
239	}
240	}
241	}	15✔
242	}
243
244	fn parse_query_expression(body: &str) -> Result<QueryExpression, Error> {	15✔
245	if body.is_empty() {	15✔
246	return Ok(QueryExpression::with_no_limit());	6✔
247	}	9✔
248		9✔
249	serde_json::from_str(body)	9✔
250	.map_err(JsonPayloadError::Deserialize)	9✔
251	.map_err(Into::into)	9✔
252	}	15✔
253	}

getdozer / dozer / 6299724219

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous