5829357348

pending completion

Build # 5829357348

Build Type

Pull #1839

github

Committed by

aaryaattrey

Commit Message

typo

Pull Request Pull Request #1839: load config from stdin

Run Details

37 of 37 new or added lines in 4 files covered. (100.0%)

45551 of 59108 relevant lines covered (77.06%)

39686.65 hits per line

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

0.0

/dozer-cli/src/cli/init.rs

use crate::errors::{CliError, OrchestrationError};
use dozer_types::constants::{DEFAULT_LAMBDAS_DIRECTORY, DEFAULT_QUERIES_DIRECTORY};
use dozer_types::log::warn;
use dozer_types::models::config::{default_cache_dir, default_home_dir, get_cache_dir};
use dozer_types::{
    constants::DEFAULT_CONFIG_PATH,
    ingestion_types::{EthConfig, EthFilter, EthLogConfig, EthProviderConfig, SnowflakeConfig},
    log::info,
    models::{
        config::Config,
        connection::{Connection, ConnectionConfig, PostgresConfig},
    },
    serde_yaml,
};
use rustyline::history::DefaultHistory;
use rustyline::{
    completion::{Completer, Pair},
    Context,
};
use rustyline::{error::ReadlineError, Editor};
use rustyline_derive::{Helper, Highlighter, Hinter, Validator};
use std::path::{Path, PathBuf};

#[derive(Helper, Highlighter, Hinter, Validator)]
pub struct InitHelper {}

impl Completer for InitHelper {
    type Candidate = Pair;
    fn complete(
        &self,
        line: &str,
        _pos: usize,
        _ctx: &Context,
    ) -> rustyline::Result<(usize, Vec<Self::Candidate>)> {
        let line = format!("{line}_");
        let mut tokens = line.split_whitespace();
        let mut last_token = String::from(tokens.next_back().unwrap());
        last_token.pop();
        let candidates: Vec<String> = vec![
            "Postgres".to_owned(),
            "Ethereum".to_owned(),
            "Snowflake".to_owned(),
        ];
        let mut match_pair: Vec<Pair> = candidates
            .iter()
            .filter_map(|f| {
                if f.to_lowercase().starts_with(&last_token.to_lowercase()) {
                    Some(Pair {
                        display: f.to_owned(),
                        replacement: f.to_owned(),
                    })
                } else {
                    None
                }
            })
            .collect();
        if match_pair.is_empty() {
            match_pair = vec![Pair {
                display: "Postgres".to_owned(),
                replacement: "Postgres".to_owned(),
            }]
        }
        Ok((line.len() - last_token.len() - 1, match_pair))
    }
}

pub fn generate_connection(connection_name: &str) -> Connection {
    match connection_name {
        "Snowflake" | "snowflake" | "S" | "s" => {
            let snowflake_config = SnowflakeConfig {
                server: "<account_name>.<region_id>.snowflakecomputing.com".to_owned(),
                port: "443".to_owned(),
                user: "bob".to_owned(),
                password: "password".to_owned(),
                database: "database".to_owned(),
                schema: "schema".to_owned(),
                warehouse: "warehouse".to_owned(),
                driver: Some("SnowflakeDSIIDriver".to_owned()),
                role: "role".to_owned(),
            };
            let connection: Connection = Connection {
                name: "snowflake".to_owned(),
                config: Some(ConnectionConfig::Snowflake(snowflake_config)),
            };
            connection
        }
        "Ethereum" | "ethereum" | "E" | "e" => {
            let eth_filter = EthFilter {
                from_block: Some(0),
                to_block: None,
                addresses: vec![],
                topics: vec![],
            };
            let ethereum_config = EthConfig {
                provider: Some(EthProviderConfig::Log(EthLogConfig {
                    wss_url: "wss://link".to_owned(),
                    filter: Some(eth_filter),
                    contracts: vec![],
                })),
            };
            let connection: Connection = Connection {
                name: "ethereum".to_owned(),
                config: Some(ConnectionConfig::Ethereum(ethereum_config)),
            };
            connection
        }
        _ => {
            let postgres_config = PostgresConfig {
                user: Some("postgres".to_owned()),
                password: Some("postgres".to_owned()),
                host: Some("localhost".to_owned()),
                port: Some(5432),
                database: Some("users".to_owned()),
                sslmode: None,
                connection_url: None,
            };
            let connection: Connection = Connection {
                name: "postgres".to_owned(),
                config: Some(ConnectionConfig::Postgres(postgres_config)),
            };
            connection
        }
    }
}
type Question = (
    String,
    Box<dyn Fn((String, &mut Config)) -> Result<(), OrchestrationError>>,
);
pub fn generate_config_repl() -> Result<(), OrchestrationError> {
    let mut rl = Editor::<InitHelper, DefaultHistory>::new()
        .map_err(|e| OrchestrationError::CliError(CliError::ReadlineError(e)))?;
    rl.set_helper(Some(InitHelper {}));
    let mut default_config = Config::default();
    let default_app_name = "quick-start-app";
    let questions: Vec<Question> = vec![
        (
            format!("question: App name ({:}): ", default_app_name),
            Box::new(move |(app_name, config)| {
                let app_name = app_name.trim();
                if app_name.is_empty() {
                    config.app_name = default_app_name.to_string();
                } else {
                    config.app_name = app_name.to_string();
                }
                Ok(())
            }),
        ),
        (
            format!("question: Home directory ({:}): ", default_home_dir()),
            Box::new(move |(home_dir, config)| {
                if home_dir.is_empty() {
                    config.home_dir = default_home_dir();
                    config.cache_dir = default_cache_dir();
                } else {
                    config.home_dir = home_dir;
                    config.cache_dir = get_cache_dir(&config.home_dir);
                }
                Ok(())
            }),
        ),
        (
            "question: Connection Type - one of: [P]ostgres, [E]thereum, [S]nowflake (Postgres): "
                .to_string(),
            Box::new(move |(connection, config)| {
                let sample_connection = generate_connection(&connection);
                config.connections.push(sample_connection);

                Ok(())
            }),
        ),
        (
            format!("question: Config path ({:}): ", DEFAULT_CONFIG_PATH),
            Box::new(move |(yaml_path, config)| {
                let mut yaml_path = yaml_path.trim();
                if yaml_path.is_empty() {
                    yaml_path = DEFAULT_CONFIG_PATH;
                }
                let f = std::fs::OpenOptions::new()
                    .create(true)
                    .write(true)
                    .open(yaml_path)
                    .map_err(|e| {
                        OrchestrationError::CliError(CliError::FileSystem(
                            yaml_path.to_string().into(),
                            e,
                        ))
                    })?;
                serde_yaml::to_writer(f, &config)
                    .map_err(OrchestrationError::FailedToWriteConfigYaml)?;

                info!("Generating workspace: \n\
                \n- {} (main configuration)\n- ./queries (folder for sql queries)\n- ./lambdas (folder for lambda functions)
                \n• More details about our config: https://getdozer.io/docs/reference/configuration/introduction\
                \n• Connector & Sources: https://getdozer.io/docs/reference/configuration/connectors\
                \n• Endpoints: https://getdozer.io/docs/reference/configuration/endpoints/",
                   yaml_path.to_owned());

                let path = PathBuf::from(yaml_path);
                if let Some(dir) = path.parent() {
                    let queries_path = Path::new(dir).join(DEFAULT_QUERIES_DIRECTORY);
                    if let Err(_e) = std::fs::create_dir(queries_path) {
                        warn!("Cannot create queries directory");
                    }

                    let lambdas_path = Path::new(dir).join(DEFAULT_LAMBDAS_DIRECTORY);
                    if let Err(_e) = std::fs::create_dir(lambdas_path) {
                        warn!("Cannot create lambdas directory");
                    }
                }

                Ok(())
            }),
        ),
    ];
    let result = questions.iter().try_for_each(|(question, func)| {
        let readline = rl.readline(question);
        match readline {
            Ok(input) => func((input, &mut default_config)),
            Err(err) => Err(OrchestrationError::CliError(CliError::ReadlineError(err))),
        }
    });

    match result {
        Ok(_) => Ok(()),
        Err(e) => match e {
            OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Interrupted)) => {
                info!("Exiting..");
                Ok(())
            }
            OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Eof)) => {
                info!("CTRL-D - exiting...");
                Ok(())
            }
            _ => Err(e),
        },
    }
}

1	use crate::errors::{CliError, OrchestrationError};
2	use dozer_types::constants::{DEFAULT_LAMBDAS_DIRECTORY, DEFAULT_QUERIES_DIRECTORY};
3	use dozer_types::log::warn;
4	use dozer_types::models::config::{default_cache_dir, default_home_dir, get_cache_dir};
5	use dozer_types::{
6	constants::DEFAULT_CONFIG_PATH,
7	ingestion_types::{EthConfig, EthFilter, EthLogConfig, EthProviderConfig, SnowflakeConfig},
8	log::info,
9	models::{
10	config::Config,
11	connection::{Connection, ConnectionConfig, PostgresConfig},
12	},
13	serde_yaml,
14	};
15	use rustyline::history::DefaultHistory;
16	use rustyline::{
17	completion::{Completer, Pair},
18	Context,
19	};
20	use rustyline::{error::ReadlineError, Editor};
21	use rustyline_derive::{Helper, Highlighter, Hinter, Validator};
22	use std::path::{Path, PathBuf};
23
24	#[derive(Helper, Highlighter, Hinter, Validator)]
25	pub struct InitHelper {}
26
27	impl Completer for InitHelper {
28	type Candidate = Pair;
29	fn complete(	×
30	&self,	×
31	line: &str,	×
32	_pos: usize,	×
33	_ctx: &Context,	×
34	) -> rustyline::Result<(usize, Vec<Self::Candidate>)> {	×
35	let line = format!("{line}_");	×
36	let mut tokens = line.split_whitespace();	×
37	let mut last_token = String::from(tokens.next_back().unwrap());	×
38	last_token.pop();	×
39	let candidates: Vec<String> = vec![	×
40	"Postgres".to_owned(),	×
41	"Ethereum".to_owned(),	×
42	"Snowflake".to_owned(),	×
43	];	×
44	let mut match_pair: Vec<Pair> = candidates	×
45	.iter()	×
46	.filter_map(\|f\| {	×
47	if f.to_lowercase().starts_with(&last_token.to_lowercase()) {	×
48	Some(Pair {	×
49	display: f.to_owned(),	×
50	replacement: f.to_owned(),	×
51	})	×
52	} else {
53	None	×
54	}
55	})	×
56	.collect();	×
57	if match_pair.is_empty() {	×
58	match_pair = vec![Pair {	×
59	display: "Postgres".to_owned(),	×
60	replacement: "Postgres".to_owned(),	×
61	}]	×
62	}	×
63	Ok((line.len() - last_token.len() - 1, match_pair))	×
64	}	×
65	}
66
67	pub fn generate_connection(connection_name: &str) -> Connection {	×
68	match connection_name {	×
69	"Snowflake" \| "snowflake" \| "S" \| "s" => {	×
70	let snowflake_config = SnowflakeConfig {	×
71	server: "<account_name>.<region_id>.snowflakecomputing.com".to_owned(),	×
72	port: "443".to_owned(),	×
73	user: "bob".to_owned(),	×
74	password: "password".to_owned(),	×
75	database: "database".to_owned(),	×
76	schema: "schema".to_owned(),	×
77	warehouse: "warehouse".to_owned(),	×
78	driver: Some("SnowflakeDSIIDriver".to_owned()),	×
79	role: "role".to_owned(),	×
80	};	×
81	let connection: Connection = Connection {	×
82	name: "snowflake".to_owned(),	×
83	config: Some(ConnectionConfig::Snowflake(snowflake_config)),	×
84	};	×
85	connection	×
86	}
87	"Ethereum" \| "ethereum" \| "E" \| "e" => {	×
88	let eth_filter = EthFilter {	×
89	from_block: Some(0),	×
90	to_block: None,	×
91	addresses: vec![],	×
92	topics: vec![],	×
93	};	×
94	let ethereum_config = EthConfig {	×
95	provider: Some(EthProviderConfig::Log(EthLogConfig {	×
96	wss_url: "wss://link".to_owned(),	×
97	filter: Some(eth_filter),	×
98	contracts: vec![],	×
99	})),	×
100	};	×
101	let connection: Connection = Connection {	×
102	name: "ethereum".to_owned(),	×
103	config: Some(ConnectionConfig::Ethereum(ethereum_config)),	×
104	};	×
105	connection	×
106	}
107	_ => {
108	let postgres_config = PostgresConfig {	×
109	user: Some("postgres".to_owned()),	×
110	password: Some("postgres".to_owned()),	×
111	host: Some("localhost".to_owned()),	×
112	port: Some(5432),	×
113	database: Some("users".to_owned()),	×
114	sslmode: None,	×
115	connection_url: None,	×
116	};	×
117	let connection: Connection = Connection {	×
118	name: "postgres".to_owned(),	×
119	config: Some(ConnectionConfig::Postgres(postgres_config)),	×
120	};	×
121	connection	×
122	}
123	}
124	}	×
125	type Question = (
126	String,
127	Box<dyn Fn((String, &mut Config)) -> Result<(), OrchestrationError>>,
128	);
129	pub fn generate_config_repl() -> Result<(), OrchestrationError> {	×
130	let mut rl = Editor::<InitHelper, DefaultHistory>::new()	×
131	.map_err(\|e\| OrchestrationError::CliError(CliError::ReadlineError(e)))?;	×
132	rl.set_helper(Some(InitHelper {}));	×
133	let mut default_config = Config::default();	×
134	let default_app_name = "quick-start-app";	×
135	let questions: Vec<Question> = vec![	×
136	(	×
137	format!("question: App name ({:}): ", default_app_name),	×
138	Box::new(move \|(app_name, config)\| {	×
139	let app_name = app_name.trim();	×
140	if app_name.is_empty() {	×
141	config.app_name = default_app_name.to_string();	×
142	} else {	×
143	config.app_name = app_name.to_string();	×
144	}	×
145	Ok(())	×
146	}),	×
147	),	×
148	(	×
149	format!("question: Home directory ({:}): ", default_home_dir()),	×
150	Box::new(move \|(home_dir, config)\| {	×
151	if home_dir.is_empty() {	×
152	config.home_dir = default_home_dir();	×
153	config.cache_dir = default_cache_dir();	×
154	} else {	×
155	config.home_dir = home_dir;	×
156	config.cache_dir = get_cache_dir(&config.home_dir);	×
157	}	×
158	Ok(())	×
159	}),	×
160	),	×
161	(	×
162	"question: Connection Type - one of: [P]ostgres, [E]thereum, [S]nowflake (Postgres): "	×
163	.to_string(),	×
164	Box::new(move \|(connection, config)\| {	×
165	let sample_connection = generate_connection(&connection);	×
166	config.connections.push(sample_connection);	×
167		×
168	Ok(())	×
169	}),	×
170	),	×
171	(	×
172	format!("question: Config path ({:}): ", DEFAULT_CONFIG_PATH),	×
173	Box::new(move \|(yaml_path, config)\| {	×
174	let mut yaml_path = yaml_path.trim();	×
175	if yaml_path.is_empty() {	×
176	yaml_path = DEFAULT_CONFIG_PATH;	×
177	}	×
178	let f = std::fs::OpenOptions::new()	×
179	.create(true)	×
180	.write(true)	×
181	.open(yaml_path)	×
182	.map_err(\|e\| {	×
183	OrchestrationError::CliError(CliError::FileSystem(	×
184	yaml_path.to_string().into(),	×
185	e,	×
186	))	×
187	})?;	×
188	serde_yaml::to_writer(f, &config)	×
189	.map_err(OrchestrationError::FailedToWriteConfigYaml)?;	×
190
191	info!("Generating workspace: \n\	×
192	\n- {} (main configuration)\n- ./queries (folder for sql queries)\n- ./lambdas (folder for lambda functions)	×
193	\n• More details about our config: https://getdozer.io/docs/reference/configuration/introduction\	×
194	\n• Connector & Sources: https://getdozer.io/docs/reference/configuration/connectors\	×
195	\n• Endpoints: https://getdozer.io/docs/reference/configuration/endpoints/",	×
196	yaml_path.to_owned());	×
197
198	let path = PathBuf::from(yaml_path);	×
199	if let Some(dir) = path.parent() {	×
200	let queries_path = Path::new(dir).join(DEFAULT_QUERIES_DIRECTORY);	×
201	if let Err(_e) = std::fs::create_dir(queries_path) {	×
202	warn!("Cannot create queries directory");	×
203	}	×
204
205	let lambdas_path = Path::new(dir).join(DEFAULT_LAMBDAS_DIRECTORY);	×
206	if let Err(_e) = std::fs::create_dir(lambdas_path) {	×
207	warn!("Cannot create lambdas directory");	×
208	}	×
209	}	×
210
211	Ok(())	×
212	}),	×
213	),	×
214	];	×
215	let result = questions.iter().try_for_each(\|(question, func)\| {	×
216	let readline = rl.readline(question);	×
217	match readline {	×
218	Ok(input) => func((input, &mut default_config)),	×
219	Err(err) => Err(OrchestrationError::CliError(CliError::ReadlineError(err))),	×
220	}
221	});	×
222		×
223	match result {	×
224	Ok(_) => Ok(()),	×
225	Err(e) => match e {	×
226	OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Interrupted)) => {
227	info!("Exiting..");	×
228	Ok(())	×
229	}
230	OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Eof)) => {
231	info!("CTRL-D - exiting...");	×
232	Ok(())	×
233	}
234	_ => Err(e),	×
235	},
236	}
237	}	×

getdozer / dozer / 5829357348

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