4698406303

Build Type

github

Committed by GitHub

Commit Message

Merge daefffe87 into b6889464a

Pull Request Pull Request #1426: feat: implement python log bindings

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

34863 of 45840 relevant lines covered (76.05%)

10764.36 hits per line

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

81.2

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

use std::sync::Arc;

// Exports
use crate::errors::ApiError;
use crate::rest::api_generator::health_route;
use crate::{
    auth::api::{auth_route, validate},
    CacheEndpoint,
};
use actix_cors::Cors;
use actix_web::{
    body::MessageBody,
    dev::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
    middleware::{Condition, Logger},
    web, App, HttpMessage, HttpServer,
};
use actix_web_httpauth::middleware::HttpAuthentication;
use dozer_types::{log::info, models::api_config::RestApiOptions};
use dozer_types::{
    models::api_security::ApiSecurity,
    serde::{self, Deserialize, Serialize},
};
use futures_util::Future;
use tracing_actix_web::TracingLogger;

mod api_generator;

pub use api_generator::field_to_json_value;

#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
#[serde(crate = "self::serde")]
enum CorsOptions {
    Permissive,
    // origins, max_age
    Custom(Vec<String>, usize),
}
#[derive(Clone)]
pub struct ApiServer {
    shutdown_timeout: u64,
    port: u16,
    cors: CorsOptions,
    security: Option<ApiSecurity>,
    host: String,
}

impl Default for ApiServer {
    fn default() -> Self {
        Self {
            shutdown_timeout: 0,
            port: 8080,
            cors: CorsOptions::Permissive,
            security: None,
            host: "0.0.0.0".to_owned(),
        }
    }
}

impl ApiServer {
    pub fn new(rest_config: RestApiOptions, security: Option<ApiSecurity>) -> Self {
        Self {
            shutdown_timeout: 0,
            port: rest_config.port as u16,
            cors: CorsOptions::Permissive,
            security,
            host: rest_config.host,
        }
    }
    fn get_cors(cors: CorsOptions) -> Cors {
        match cors {
            CorsOptions::Permissive => Cors::permissive(),
            CorsOptions::Custom(origins, max_age) => origins
                .into_iter()
                .fold(Cors::default(), |cors, origin| cors.allowed_origin(&origin))
                .max_age(max_age),
        }
    }

    fn create_app_entry(
        security: Option<ApiSecurity>,
        cors: CorsOptions,
        cache_endpoints: Vec<Arc<CacheEndpoint>>,
    ) -> App<
        impl ServiceFactory<
            ServiceRequest,
            Response = ServiceResponse<impl MessageBody>,
            Config = (),
            InitError = (),
            Error = actix_web::Error,
        >,
    > {
        let mut app = App::new()
            .wrap(Logger::default())
            .wrap(TracingLogger::default());

        let is_auth_configured = if let Some(api_security) = security {
            // Injecting API Security
            app = app.app_data(api_security);
            true
        } else {
            false
        };
        let auth_middleware =
            Condition::new(is_auth_configured, HttpAuthentication::bearer(validate));

        let cors_middleware = Self::get_cors(cors);

        cache_endpoints
            .into_iter()
            .fold(app, |app, cache_endpoint| {
                let endpoint = &cache_endpoint.endpoint;
                let scope = &endpoint.path;
                app.service(
                    web::scope(scope)
                        // Inject cache_endpoint for generated functions
                        .wrap_fn(move |req, srv| {
                            req.extensions_mut().insert(cache_endpoint.clone());
                            srv.call(req)
                        })
                        .route("/count", web::post().to(api_generator::count))
                        .route("/query", web::post().to(api_generator::query))
                        .route("/oapi", web::post().to(api_generator::generate_oapi))
                        .route("/{id}", web::get().to(api_generator::get))
                        .route("/", web::get().to(api_generator::list))
                        .route("", web::get().to(api_generator::list)),
                )
            })
            // Attach token generation route
            .route("/auth/token", web::post().to(auth_route))
            // Attach health route
            .route("/health", web::get().to(health_route))
            // Wrap Api Validator
            .wrap(auth_middleware)
            // Wrap CORS around api validator. Required to return the right headers.
            .wrap(cors_middleware)
    }

    pub async fn run(
        self,
        cache_endpoints: Vec<Arc<CacheEndpoint>>,
        shutdown: impl Future<Output = ()> + Send + 'static,
    ) -> Result<(), ApiError> {
        info!(
            "Starting Rest Api Server on http://{}:{} with security: {}",
            self.host,
            self.port,
            self.security
                .as_ref()
                .map_or("None".to_string(), |s| match s {
                    ApiSecurity::Jwt(_) => "JWT".to_string(),
                })
        );
        let cors = self.cors;
        let security = self.security;
        let address = format!("{}:{}", self.host, self.port);
        let server = HttpServer::new(move || {
            ApiServer::create_app_entry(security.clone(), cors.clone(), cache_endpoints.clone())
        })
        .bind(&address)
        .map_err(|e| ApiError::FailedToBindToAddress(address, e))?
        .disable_signals()
        .shutdown_timeout(self.shutdown_timeout)
        .run();

        let server_handle = server.handle();
        tokio::spawn(async move {
            shutdown.await;
            server_handle.stop(true).await;
        });

        server
            .await
            .map_err(|e| ApiError::InternalError(Box::new(e)))
    }
}

#[cfg(test)]
mod tests;

1	use std::sync::Arc;
2
3	// Exports
4	use crate::errors::ApiError;
5	use crate::rest::api_generator::health_route;
6	use crate::{
7	auth::api::{auth_route, validate},
8	CacheEndpoint,
9	};
10	use actix_cors::Cors;
11	use actix_web::{
12	body::MessageBody,
13	dev::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
14	middleware::{Condition, Logger},
15	web, App, HttpMessage, HttpServer,
16	};
17	use actix_web_httpauth::middleware::HttpAuthentication;
18	use dozer_types::{log::info, models::api_config::RestApiOptions};
19	use dozer_types::{
20	models::api_security::ApiSecurity,
21	serde::{self, Deserialize, Serialize},
22	};
23	use futures_util::Future;
24	use tracing_actix_web::TracingLogger;
25
26	mod api_generator;
27
28	pub use api_generator::field_to_json_value;
29
30	#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]	204✔
31	#[serde(crate = "self::serde")]
32	enum CorsOptions {
33	Permissive,
34	// origins, max_age
35	Custom(Vec<String>, usize),
36	}
37	#[derive(Clone)]	×
38	pub struct ApiServer {
39	shutdown_timeout: u64,
40	port: u16,
41	cors: CorsOptions,
42	security: Option<ApiSecurity>,
43	host: String,
44	}
45
46	impl Default for ApiServer {
47	fn default() -> Self {	×
48	Self {	×
49	shutdown_timeout: 0,	×
50	port: 8080,	×
51	cors: CorsOptions::Permissive,	×
52	security: None,	×
53	host: "0.0.0.0".to_owned(),	×
54	}	×
55	}	×
56	}
57
58	impl ApiServer {
59	pub fn new(rest_config: RestApiOptions, security: Option<ApiSecurity>) -> Self {	15✔
60	Self {	15✔
61	shutdown_timeout: 0,	15✔
62	port: rest_config.port as u16,	15✔
63	cors: CorsOptions::Permissive,	15✔
64	security,	15✔
65	host: rest_config.host,	15✔
66	}	15✔
67	}	15✔
68	fn get_cors(cors: CorsOptions) -> Cors {	128✔
69	match cors {	128✔
70	CorsOptions::Permissive => Cors::permissive(),	128✔
71	CorsOptions::Custom(origins, max_age) => origins	×
72	.into_iter()	×
73	.fold(Cors::default(), \|cors, origin\| cors.allowed_origin(&origin))	×
74	.max_age(max_age),	×
75	}
76	}	128✔
77
78	fn create_app_entry(	128✔
79	security: Option<ApiSecurity>,	128✔
80	cors: CorsOptions,	128✔
81	cache_endpoints: Vec<Arc<CacheEndpoint>>,	128✔
82	) -> App<	128✔
83	impl ServiceFactory<	128✔
84	ServiceRequest,	128✔
85	Response = ServiceResponse<impl MessageBody>,	128✔
86	Config = (),	128✔
87	InitError = (),	128✔
88	Error = actix_web::Error,	128✔
89	>,	128✔
90	> {	128✔
91	let mut app = App::new()	128✔
92	.wrap(Logger::default())	128✔
93	.wrap(TracingLogger::default());	128✔
94
95	let is_auth_configured = if let Some(api_security) = security {	128✔
96	// Injecting API Security
97	app = app.app_data(api_security);	4✔
98	true	4✔
99	} else {
100	false	124✔
101	};
102	let auth_middleware =	128✔
103	Condition::new(is_auth_configured, HttpAuthentication::bearer(validate));	128✔
104		128✔
105	let cors_middleware = Self::get_cors(cors);	128✔
106		128✔
107	cache_endpoints	128✔
108	.into_iter()	128✔
109	.fold(app, \|app, cache_endpoint\| {	128✔
110	let endpoint = &cache_endpoint.endpoint;	128✔
111	let scope = &endpoint.path;	128✔
112	app.service(	128✔
113	web::scope(scope)	128✔
114	// Inject cache_endpoint for generated functions	128✔
115	.wrap_fn(move \|req, srv\| {	134✔
116	req.extensions_mut().insert(cache_endpoint.clone());	14✔
117	srv.call(req)	14✔
118	})	134✔
119	.route("/count", web::post().to(api_generator::count))	128✔
120	.route("/query", web::post().to(api_generator::query))	128✔
121	.route("/oapi", web::post().to(api_generator::generate_oapi))	128✔
122	.route("/{id}", web::get().to(api_generator::get))	128✔
123	.route("/", web::get().to(api_generator::list))	128✔
124	.route("", web::get().to(api_generator::list)),	128✔
125	)	128✔
126	})	128✔
127	// Attach token generation route	128✔
128	.route("/auth/token", web::post().to(auth_route))	128✔
129	// Attach health route	128✔
130	.route("/health", web::get().to(health_route))	128✔
131	// Wrap Api Validator	128✔
132	.wrap(auth_middleware)	128✔
133	// Wrap CORS around api validator. Required to return the right headers.	128✔
134	.wrap(cors_middleware)	128✔
135	}	128✔
136
137	pub async fn run(	12✔
138	self,	12✔
139	cache_endpoints: Vec<Arc<CacheEndpoint>>,	12✔
140	shutdown: impl Future<Output = ()> + Send + 'static,	12✔
141	) -> Result<(), ApiError> {	12✔
142	info!(
143	"Starting Rest Api Server on http://{}:{} with security: {}",	×
144	self.host,	×
145	self.port,	×
146	self.security	×
147	.as_ref()	×
148	.map_or("None".to_string(), \|s\| match s {	×
149	ApiSecurity::Jwt(_) => "JWT".to_string(),	×
150	})	×
151	);
152	let cors = self.cors;	12✔
153	let security = self.security;	12✔
154	let address = format!("{}:{}", self.host, self.port);	12✔
155	let server = HttpServer::new(move \|\| {	96✔
156	ApiServer::create_app_entry(security.clone(), cors.clone(), cache_endpoints.clone())	96✔
157	})	96✔
158	.bind(&address)	12✔
159	.map_err(\|e\| ApiError::FailedToBindToAddress(address, e))?	12✔
160	.disable_signals()	12✔
161	.shutdown_timeout(self.shutdown_timeout)	12✔
162	.run();	12✔
163		12✔
164	let server_handle = server.handle();	12✔
165	tokio::spawn(async move {	12✔
166	shutdown.await;	12✔
167	server_handle.stop(true).await;	12✔
168	});	12✔
169		12✔
170	server	12✔
171	.await	36✔
172	.map_err(\|e\| ApiError::InternalError(Box::new(e)))	12✔
173	}	12✔
174	}
175
176	#[cfg(test)]
177	mod tests;

getdozer / dozer / 4698406303

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