4370280743

Build Type

push

github

Committed by GitHub

Commit Message

Bump async-trait from 0.1.65 to 0.1.66 (#1179)

Run Details

27808 of 38702 relevant lines covered (71.85%)

25323.55 hits per line

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

67.8

/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::{default_limit_for_query, QueryExpression, Skip};
use dozer_cache::cache::{index, RecordWithId};
use dozer_cache::CacheReader;
use dozer_types::chrono::SecondsFormat;
use dozer_types::errors::types::TypeError;
use dozer_types::indexmap::IndexMap;
use dozer_types::log::info;
use dozer_types::models::api_endpoint::ApiEndpoint;
use dozer_types::ordered_float::OrderedFloat;
use dozer_types::types::{Field, Schema, DATE_FORMAT};
use openapiv3::OpenAPI;

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

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

    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<RoCacheEndpoint>>,
) -> 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<RoCacheEndpoint>>,
    path: web::Path<String>,
) -> Result<HttpResponse, ApiError> {
    let cache_reader = &cache_endpoint.cache_reader();
    let schema = &cache_reader
        .get_schema()
        .map_err(ApiError::SchemaNotFound)?
        .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)?
    } else {
        return Err(ApiError::MultiIndexFetch(key.to_string()));
    };

    let key = index::get_primary_key(&[0], &[key]);
    let record = get_record(
        &cache_endpoint.cache_reader(),
        &key,
        access.map(|a| a.into_inner()),
    )?;

    Ok(record_to_map(record, schema).map(|map| HttpResponse::Ok().json(map))?)
}

// Generated list function for multiple records with a default query expression
pub async fn list(
    access: Option<ReqData<Access>>,
    cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,
) -> Result<HttpResponse, ApiError> {
    let mut exp = QueryExpression::new(None, vec![], Some(50), Skip::Skip(0));
    match get_records_map(access, cache_endpoint, &mut exp) {
        Ok(maps) => Ok(HttpResponse::Ok().json(maps)),
        Err(e) => match e {
            ApiError::QueryFailed(_) => {
                let res: Vec<String> = vec![];
                info!("No records found.");
                Ok(HttpResponse::Ok().json(res))
            }
            _ => Err(ApiError::InternalError(Box::new(e))),
        },
    }
}

// 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<RoCacheEndpoint>>,
    query_info: Option<web::Json<Value>>,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = match query_info {
        Some(query_info) => serde_json::from_value::<QueryExpression>(query_info.0)
            .map_err(ApiError::map_deserialization_error)?,
        None => QueryExpression::with_no_limit(),
    };

    get_records_count(
        &cache_endpoint.cache_reader(),
        &mut query_expression,
        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<RoCacheEndpoint>>,
    query_info: Option<web::Json<Value>>,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = match query_info {
        Some(query_info) => serde_json::from_value::<QueryExpression>(query_info.0)
            .map_err(ApiError::map_deserialization_error)?,
        None => QueryExpression::with_default_limit(),
    };
    if query_expression.limit.is_none() {
        query_expression.limit = Some(default_limit_for_query());
    }

    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<RoCacheEndpoint>>,
    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, access.map(|a| a.into_inner()))?;
    let schema = &cache_reader
        .get_schema()
        .map_err(ApiError::SchemaNotFound)?
        .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: RecordWithId,
    schema: &Schema,
) -> Result<IndexMap<String, Value>, TypeError> {
    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.record.version),
    );

    Ok(map)
}

fn convert_x_y_to_object((x, y): &(OrderedFloat<f64>, OrderedFloat<f64>)) -> Value {
    let mut m = Map::new();
    m.insert("x".to_string(), Value::from(x.0));
    m.insert("y".to_string(), Value::from(y.0));
    Value::Object(m)
}

/// Used in REST APIs for converting raw value back and forth.
///
/// Should be consistent with `convert_cache_type_to_schema_type`.
fn field_to_json_value(field: Field) -> Value {
    match field {
        Field::UInt(n) => Value::from(n),
        Field::Int(n) => Value::from(n),
        Field::Float(n) => Value::from(n.0),
        Field::Boolean(b) => Value::from(b),
        Field::String(s) => Value::from(s),
        Field::Text(n) => Value::from(n),
        Field::Binary(b) => Value::from(b),
        Field::Decimal(n) => Value::String(n.to_string()),
        Field::Timestamp(ts) => Value::String(ts.to_rfc3339_opts(SecondsFormat::Millis, true)),
        Field::Date(n) => Value::String(n.format(DATE_FORMAT).to_string()),
        Field::Bson(b) => Value::from(b),
        Field::Point(point) => convert_x_y_to_object(&point.0.x_y()),
        Field::Null => Value::Null,
    }
}

#[cfg(test)]
mod tests {
    use dozer_types::{
        chrono::{NaiveDate, Offset, TimeZone, Utc},
        json_value_to_field,
        ordered_float::OrderedFloat,
        rust_decimal::Decimal,
        types::{DozerPoint, Field, FieldType},
    };

    use super::*;

    fn test_field_conversion(field_type: FieldType, field: Field) {
        // Convert the field to a JSON value.
        let value = field_to_json_value(field.clone());

        // Convert the JSON value back to a Field.
        let deserialized = json_value_to_field(value, field_type, true).unwrap();

        assert_eq!(deserialized, field, "must be equal");
    }

    #[test]
    fn test_field_types_json_conversion() {
        let fields = vec![
            (FieldType::Int, Field::Int(-1)),
            (FieldType::UInt, Field::UInt(1)),
            (FieldType::Float, Field::Float(OrderedFloat(1.1))),
            (FieldType::Boolean, Field::Boolean(true)),
            (FieldType::String, Field::String("a".to_string())),
            (FieldType::Binary, Field::Binary(b"asdf".to_vec())),
            (FieldType::Decimal, Field::Decimal(Decimal::new(202, 2))),
            (
                FieldType::Timestamp,
                Field::Timestamp(Utc.fix().with_ymd_and_hms(2001, 1, 1, 0, 4, 0).unwrap()),
            ),
            (
                FieldType::Date,
                Field::Date(NaiveDate::from_ymd_opt(2022, 11, 24).unwrap()),
            ),
            (
                FieldType::Bson,
                Field::Bson(vec![
                    // BSON representation of `{"abc":"foo"}`
                    123, 34, 97, 98, 99, 34, 58, 34, 102, 111, 111, 34, 125,
                ]),
            ),
            (FieldType::Text, Field::Text("lorem ipsum".to_string())),
            (
                FieldType::Point,
                Field::Point(DozerPoint::from((3.234, 4.567))),
            ),
        ];
        for (field_type, field) in fields {
            test_field_conversion(field_type, field);
        }
    }
}

1	use std::sync::Arc;
2
3	use actix_web::web::ReqData;
4	use actix_web::{web, HttpResponse};
5	use dozer_cache::cache::expression::{default_limit_for_query, QueryExpression, Skip};
6	use dozer_cache::cache::{index, RecordWithId};
7	use dozer_cache::CacheReader;
8	use dozer_types::chrono::SecondsFormat;
9	use dozer_types::errors::types::TypeError;
10	use dozer_types::indexmap::IndexMap;
11	use dozer_types::log::info;
12	use dozer_types::models::api_endpoint::ApiEndpoint;
13	use dozer_types::ordered_float::OrderedFloat;
14	use dozer_types::types::{Field, Schema, DATE_FORMAT};
15	use openapiv3::OpenAPI;
16
17	use crate::api_helper::{get_record, get_records, get_records_count};
18	use crate::generator::oapi::generator::OpenApiGenerator;
19	use crate::RoCacheEndpoint;
20	use crate::{auth::Access, errors::ApiError};
21	use dozer_types::grpc_types::health::health_check_response::ServingStatus;
22	use dozer_types::serde_json;
23	use dozer_types::serde_json::{json, Map, Value};
24
25	fn generate_oapi3(reader: &CacheReader, endpoint: ApiEndpoint) -> Result<OpenAPI, ApiError> {	×
26	let (schema, secondary_indexes) = reader.get_schema().map_err(ApiError::SchemaNotFound)?;	×
27		×
28	let oapi_generator = OpenApiGenerator::new(	×
29	schema,	×
30	secondary_indexes,	×
31	endpoint,	×
32	vec![format!("http://localhost:{}", "8080")],	×
33	);	×
34		×
35	Ok(oapi_generator.generate_oas3())	×
36	}	×
37		×
38	/// Generated function to return openapi.yaml documentation.	×
39	pub async fn generate_oapi(	×
40	cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,	×
41	) -> Result<HttpResponse, ApiError> {	×
42	generate_oapi3(	×
43	&cache_endpoint.cache_reader(),	×
44	cache_endpoint.endpoint.clone(),	×
45	)	×
46	.map(\|result\| HttpResponse::Ok().json(result))	×
47	}	×
48		×
49	// Generated Get function to return a single record in JSON format	×
50	pub async fn get(	1✔
51	access: Option<ReqData<Access>>,	1✔
52	cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,	1✔
53	path: web::Path<String>,	1✔
54	) -> Result<HttpResponse, ApiError> {	1✔
55	let cache_reader = &cache_endpoint.cache_reader();	1✔
56	let schema = &cache_reader	1✔
57	.get_schema()	1✔
58	.map_err(ApiError::SchemaNotFound)?	1✔
59	.0;	×
60		×
61	let key = path.as_str();	1✔
62	let key = if schema.primary_index.is_empty() {	1✔
63	return Err(ApiError::NoPrimaryKey);	×
64	} else if schema.primary_index.len() == 1 {	1✔
65	let field = &schema.fields[schema.primary_index[0]];	1✔
66	Field::from_str(key, field.typ, field.nullable)?	1✔
67	} else {	×
68	return Err(ApiError::MultiIndexFetch(key.to_string()));	×
69	};
70		×
71	let key = index::get_primary_key(&[0], &[key]);	1✔
72	let record = get_record(	1✔
73	&cache_endpoint.cache_reader(),	1✔
74	&key,	1✔
75	access.map(\|a\| a.into_inner()),	1✔
76	)?;	1✔
77		×
78	Ok(record_to_map(record, schema).map(\|map\| HttpResponse::Ok().json(map))?)	1✔
79	}	1✔
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<RoCacheEndpoint>>,	3✔
85	) -> Result<HttpResponse, ApiError> {	3✔
86	let mut exp = QueryExpression::new(None, vec![], Some(50), Skip::Skip(0));	3✔
87	match get_records_map(access, cache_endpoint, &mut exp) {	3✔
88	Ok(maps) => Ok(HttpResponse::Ok().json(maps)),	3✔
89	Err(e) => match e {	×
90	ApiError::QueryFailed(_) => {	×
91	let res: Vec<String> = vec![];	×
92	info!("No records found.");	×
93	Ok(HttpResponse::Ok().json(res))	×
94	}	×
95	_ => Err(ApiError::InternalError(Box::new(e))),	×
96	},
97	}	×
98	}	3✔
99
100	// Generated get function for health check	×
101	pub async fn health_route() -> Result<HttpResponse, ApiError> {	×
102	let status = ServingStatus::Serving;	×
103	let resp = json!({ "status": status.as_str_name() }).to_string();	×
104	Ok(HttpResponse::Ok().body(resp))	×
105	}	×
106		×
107	pub async fn count(	5✔
108	access: Option<ReqData<Access>>,	5✔
109	cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,	5✔
110	query_info: Option<web::Json<Value>>,	5✔
111	) -> Result<HttpResponse, ApiError> {	5✔
112	let mut query_expression = match query_info {	5✔
113	Some(query_info) => serde_json::from_value::<QueryExpression>(query_info.0)	4✔
114	.map_err(ApiError::map_deserialization_error)?,	4✔
115	None => QueryExpression::with_no_limit(),	1✔
116	};	×
117		×
118	get_records_count(	5✔
119	&cache_endpoint.cache_reader(),	5✔
120	&mut query_expression,	5✔
121	access.map(\|a\| a.into_inner()),	5✔
122	)	5✔
123	.map(\|count\| HttpResponse::Ok().json(count))	5✔
124	}	5✔
125		×
126	// Generated query function for multiple records	×
127	pub async fn query(	5✔
128	access: Option<ReqData<Access>>,	5✔
129	cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,	5✔
130	query_info: Option<web::Json<Value>>,	5✔
131	) -> Result<HttpResponse, ApiError> {	5✔
132	let mut query_expression = match query_info {	5✔
133	Some(query_info) => serde_json::from_value::<QueryExpression>(query_info.0)	4✔
134	.map_err(ApiError::map_deserialization_error)?,	4✔
135	None => QueryExpression::with_default_limit(),	1✔
136	};	×
137	if query_expression.limit.is_none() {	5✔
138	query_expression.limit = Some(default_limit_for_query());	3✔
139	}	3✔
140		×
141	get_records_map(access, cache_endpoint, &mut query_expression)	5✔
142	.map(\|maps\| HttpResponse::Ok().json(maps))	5✔
143	}	5✔
144		×
145	/// Get multiple records	×
146	fn get_records_map(	8✔
147	access: Option<ReqData<Access>>,	8✔
148	cache_endpoint: ReqData<Arc<RoCacheEndpoint>>,	8✔
149	exp: &mut QueryExpression,	8✔
150	) -> Result<Vec<IndexMap<String, Value>>, ApiError> {	8✔
151	let mut maps = vec![];	8✔
152	let cache_reader = &cache_endpoint.cache_reader();	8✔
153	let records = get_records(cache_reader, exp, access.map(\|a\| a.into_inner()))?;	8✔
154	let schema = &cache_reader	8✔
155	.get_schema()	8✔
156	.map_err(ApiError::SchemaNotFound)?	8✔
157	.0;	×
158	for record in records.into_iter() {	312✔
159	let map = record_to_map(record, schema)?;	312✔
160	maps.push(map);	312✔
161	}	×
162	Ok(maps)	8✔
163	}	8✔
164		×
165	/// Used in REST APIs for converting to JSON
166	fn record_to_map(	313✔
167	record: RecordWithId,	313✔
168	schema: &Schema,	313✔
169	) -> Result<IndexMap<String, Value>, TypeError> {	313✔
170	let mut map = IndexMap::new();	313✔
171		×
172	for (field_def, field) in schema.fields.iter().zip(record.record.values) {	1,565✔
173	let val = field_to_json_value(field);	1,565✔
174	map.insert(field_def.name.clone(), val);	1,565✔
175	}	1,565✔
176		×
177	map.insert("__dozer_record_id".to_string(), Value::from(record.id));	313✔
178	map.insert(	313✔
179	"__dozer_record_version".to_string(),	313✔
180	Value::from(record.record.version),	313✔
181	);	313✔
182		313✔
183	Ok(map)	313✔
184	}	313✔
185		×
186	fn convert_x_y_to_object((x, y): &(OrderedFloat<f64>, OrderedFloat<f64>)) -> Value {	1✔
187	let mut m = Map::new();	1✔
188	m.insert("x".to_string(), Value::from(x.0));	1✔
189	m.insert("y".to_string(), Value::from(y.0));	1✔
190	Value::Object(m)	1✔
191	}	1✔
192		×
193	/// Used in REST APIs for converting raw value back and forth.	×
194	///	×
195	/// Should be consistent with `convert_cache_type_to_schema_type`.	×
196	fn field_to_json_value(field: Field) -> Value {	1,577✔
197	match field {	1,577✔
198	Field::UInt(n) => Value::from(n),	627✔
199	Field::Int(n) => Value::from(n),	1✔
200	Field::Float(n) => Value::from(n.0),	1✔
201	Field::Boolean(b) => Value::from(b),	1✔
202	Field::String(s) => Value::from(s),	314✔
203	Field::Text(n) => Value::from(n),	1✔
204	Field::Binary(b) => Value::from(b),	1✔
205	Field::Decimal(n) => Value::String(n.to_string()),	1✔
206	Field::Timestamp(ts) => Value::String(ts.to_rfc3339_opts(SecondsFormat::Millis, true)),	1✔
207	Field::Date(n) => Value::String(n.format(DATE_FORMAT).to_string()),	1✔
208	Field::Bson(b) => Value::from(b),	1✔
209	Field::Point(point) => convert_x_y_to_object(&point.0.x_y()),	1✔
210	Field::Null => Value::Null,	626✔
211	}	×
212	}	1,577✔
213		×
214	#[cfg(test)]	×
215	mod tests {
216	use dozer_types::{	×
217	chrono::{NaiveDate, Offset, TimeZone, Utc},
218	json_value_to_field,
219	ordered_float::OrderedFloat,
220	rust_decimal::Decimal,
221	types::{DozerPoint, Field, FieldType},
222	};
223
224	use super::*;
225
226	fn test_field_conversion(field_type: FieldType, field: Field) {	12✔
227	// Convert the field to a JSON value.	12✔
228	let value = field_to_json_value(field.clone());	12✔
229		12✔
230	// Convert the JSON value back to a Field.	12✔
231	let deserialized = json_value_to_field(value, field_type, true).unwrap();	12✔
232		12✔
233	assert_eq!(deserialized, field, "must be equal");	12✔
234	}	12✔
235		×
236	#[test]	1✔
237	fn test_field_types_json_conversion() {	1✔
238	let fields = vec![	1✔
239	(FieldType::Int, Field::Int(-1)),	1✔
240	(FieldType::UInt, Field::UInt(1)),	1✔
241	(FieldType::Float, Field::Float(OrderedFloat(1.1))),	1✔
242	(FieldType::Boolean, Field::Boolean(true)),	1✔
243	(FieldType::String, Field::String("a".to_string())),	1✔
244	(FieldType::Binary, Field::Binary(b"asdf".to_vec())),	1✔
245	(FieldType::Decimal, Field::Decimal(Decimal::new(202, 2))),	1✔
246	(	1✔
247	FieldType::Timestamp,	1✔
248	Field::Timestamp(Utc.fix().with_ymd_and_hms(2001, 1, 1, 0, 4, 0).unwrap()),	1✔
249	),	1✔
250	(	1✔
251	FieldType::Date,	1✔
252	Field::Date(NaiveDate::from_ymd_opt(2022, 11, 24).unwrap()),	1✔
253	),	1✔
254	(	1✔
255	FieldType::Bson,	1✔
256	Field::Bson(vec![	1✔
257	// BSON representation of `{"abc":"foo"}`	1✔
258	123, 34, 97, 98, 99, 34, 58, 34, 102, 111, 111, 34, 125,	1✔
259	]),	1✔
260	),	1✔
261	(FieldType::Text, Field::Text("lorem ipsum".to_string())),	1✔
262	(	1✔
263	FieldType::Point,	1✔
264	Field::Point(DozerPoint::from((3.234, 4.567))),	1✔
265	),	1✔
266	];	1✔
267	for (field_type, field) in fields {	13✔
268	test_field_conversion(field_type, field);	12✔
269	}	12✔
270	}	1✔
271	}	×

getdozer / dozer / 4370280743

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