5709656380

pending completion

Build # 5709656380

Build Type

push

github

Committed by

web-flow

Commit Message

Version bump (#1808)

Run Details

45512 of 59772 relevant lines covered (76.14%)

39312.43 hits per line

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

66.03

/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::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};

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_info: Option<web::Json<QueryExpression>>,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = match query_info {
        Some(query_info) => query_info.0,
        None => QueryExpression::with_no_limit(),
    };

    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_info: Option<web::Json<QueryExpression>>,
) -> Result<HttpResponse, ApiError> {
    let mut query_expression = match query_info {
        Some(query_info) => query_info.0,
        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<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))
}

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::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	fn generate_oapi3(reader: &CacheReader, endpoint: ApiEndpoint) -> Result<OpenAPI, ApiError> {	×
23	let (schema, secondary_indexes) = reader.get_schema();	×
24		×
25	let oapi_generator = OpenApiGenerator::new(	×
26	schema,	×
27	secondary_indexes,	×
28	endpoint,	×
29	vec![format!("http://localhost:{}", "8080")],	×
30	);	×
31		×
32	Ok(oapi_generator.generate_oas3())	×
33	}	×
34		×
35	/// Generated function to return openapi.yaml documentation.
36	pub async fn generate_oapi(	×
37	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	×
38	) -> Result<HttpResponse, ApiError> {	×
39	generate_oapi3(	×
40	&cache_endpoint.cache_reader(),	×
41	cache_endpoint.endpoint.clone(),	×
42	)	×
43	.map(\|result\| HttpResponse::Ok().json(result))	×
44	}	×
45		×
46	// Generated Get function to return a single record in JSON format
47	pub async fn get(	2✔
48	access: Option<ReqData<Access>>,	2✔
49	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	2✔
50	path: web::Path<String>,	2✔
51	) -> Result<HttpResponse, ApiError> {	2✔
52	let cache_reader = &cache_endpoint.cache_reader();	2✔
53	let schema = &cache_reader.get_schema().0;	2✔
54		2✔
55	let key = path.as_str();	2✔
56	let key = if schema.primary_index.is_empty() {	2✔
57	return Err(ApiError::NoPrimaryKey);	×
58	} else if schema.primary_index.len() == 1 {	2✔
59	let field = &schema.fields[schema.primary_index[0]];	2✔
60	Field::from_str(key, field.typ, field.nullable).map_err(ApiError::InvalidPrimaryKey)?	2✔
61	} else {	×
62	return Err(ApiError::MultiIndexFetch(key.to_string()));	×
63	};	×
64
65	// This implementation must be consistent with `dozer_cache::cache::index::get_primary_key`
66	let key = key.encode();	1✔
67	let record = get_record(	1✔
68	&cache_endpoint.cache_reader(),	1✔
69	&key,	1✔
70	&cache_endpoint.endpoint.name,	1✔
71	access.map(\|a\| a.into_inner()),	1✔
72	)?;	1✔
73		×
74	record_to_map(record, schema)	1✔
75	.map(\|map\| HttpResponse::Ok().json(map))	1✔
76	.map_err(Into::into)	1✔
77	}	2✔
78
79	// Generated list function for multiple records with a default query expression	×
80	pub async fn list(	3✔
81	access: Option<ReqData<Access>>,	3✔
82	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	3✔
83	) -> Result<HttpResponse, ApiError> {	3✔
84	let mut exp = QueryExpression::new(None, vec![], Some(50), Skip::Skip(0));	3✔
85	get_records_map(access, cache_endpoint, &mut exp).map(\|map\| HttpResponse::Ok().json(map))	3✔
86	}	3✔
87
88	// Generated get function for health check	×
89	pub async fn health_route() -> Result<HttpResponse, ApiError> {	×
90	let status = ServingStatus::Serving;	×
91	let resp = json!({ "status": status.as_str_name() }).to_string();	×
92	Ok(HttpResponse::Ok().body(resp))	×
93	}	×
94
95	pub async fn count(	5✔
96	access: Option<ReqData<Access>>,	5✔
97	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	5✔
98	query_info: Option<web::Json<QueryExpression>>,	5✔
99	) -> Result<HttpResponse, ApiError> {	5✔
100	let mut query_expression = match query_info {	5✔
101	Some(query_info) => query_info.0,	4✔
102	None => QueryExpression::with_no_limit(),	1✔
103	};
104		×
105	get_records_count(	5✔
106	&cache_endpoint.cache_reader(),	5✔
107	&mut query_expression,	5✔
108	&cache_endpoint.endpoint.name,	5✔
109	access.map(\|a\| a.into_inner()),	5✔
110	)	5✔
111	.map(\|count\| HttpResponse::Ok().json(count))	5✔
112	}	5✔
113
114	// Generated query function for multiple records	×
115	pub async fn query(	5✔
116	access: Option<ReqData<Access>>,	5✔
117	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	5✔
118	query_info: Option<web::Json<QueryExpression>>,	5✔
119	) -> Result<HttpResponse, ApiError> {	5✔
120	let mut query_expression = match query_info {	5✔
121	Some(query_info) => query_info.0,	4✔
122	None => QueryExpression::with_default_limit(),	1✔
123	};
124	if query_expression.limit.is_none() {	5✔
125	query_expression.limit = Some(default_limit_for_query());	3✔
126	}	3✔
127		×
128	get_records_map(access, cache_endpoint, &mut query_expression)	5✔
129	.map(\|maps\| HttpResponse::Ok().json(maps))	5✔
130	}	5✔
131		×
132	/// Get multiple records
133	fn get_records_map(	8✔
134	access: Option<ReqData<Access>>,	8✔
135	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	8✔
136	exp: &mut QueryExpression,	8✔
137	) -> Result<Vec<IndexMap<String, Value>>, ApiError> {	8✔
138	let mut maps = vec![];	8✔
139	let cache_reader = &cache_endpoint.cache_reader();	8✔
140	let records = get_records(	8✔
141	cache_reader,	8✔
142	exp,	8✔
143	&cache_endpoint.endpoint.name,	8✔
144	access.map(\|a\| a.into_inner()),	8✔
145	)?;	8✔
146	let schema = &cache_reader.get_schema().0;	8✔
147	for record in records.into_iter() {	312✔
148	let map = record_to_map(record, schema)?;	312✔
149	maps.push(map);	312✔
150	}	×
151	Ok(maps)	8✔
152	}	8✔
153		×
154	/// Used in REST APIs for converting to JSON	×
155	fn record_to_map(	313✔
156	record: CacheRecord,	313✔
157	schema: &Schema,	313✔
158	) -> Result<IndexMap<String, Value>, CannotConvertF64ToJson> {	313✔
159	let mut map = IndexMap::new();	313✔
160		×
161	for (field_def, field) in schema.fields.iter().zip(record.record.values) {	1,565✔
162	let val = field_to_json_value(field)?;	1,565✔
163	map.insert(field_def.name.clone(), val);	1,565✔
164	}	×
165		×
166	map.insert("__dozer_record_id".to_string(), Value::from(record.id));	313✔
167	map.insert(	313✔
168	"__dozer_record_version".to_string(),	313✔
169	Value::from(record.version),	313✔
170	);	313✔
171		313✔
172	Ok(map)	313✔
173	}	313✔
174
175	pub async fn get_phase(	1✔
176	cache_endpoint: ReqData<Arc<CacheEndpoint>>,	1✔
177	) -> Result<web::Json<Phase>, ApiError> {	1✔
178	let cache_reader = cache_endpoint.cache_reader();	1✔
179	let phase = cache_reader.get_phase().map_err(ApiError::GetPhaseFailed)?;	1✔
180	Ok(web::Json(phase))	1✔
181	}	1✔

getdozer / dozer / 5709656380

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