4295401807

Build Type

push

github

Committed by GitHub

Commit Message

Bump version (#1099)

Run Details

28685 of 39545 relevant lines covered (72.54%)

52105.29 hits per line

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

74.43

/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_and_indexes_by_name(&endpoint.name)
        .map_err(ApiError::SchemaNotFound)?;

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

    oapi_generator
        .generate_oas3()
        .map_err(ApiError::ApiGenerationError)
}

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

getdozer / dozer / 4295401807

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