-
Notifications
You must be signed in to change notification settings - Fork 216
/
rpc.rs
107 lines (89 loc) 路 4.04 KB
/
rpc.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
use crate::command_error::CommandError;
use crate::error::Error;
use crate::error_rendering::render_jsonrpc_error;
use futures::{FutureExt, TryFutureExt};
use introspection_connector::{DatabaseMetadata, IntrospectionConnector};
use jsonrpc_derive::rpc;
use serde_derive::*;
use sql_introspection_connector::SqlIntrospectionConnector;
type RpcError = jsonrpc_core::Error;
type RpcResult<T> = Result<T, RpcError>;
type RpcFutureResult<T> = Box<dyn futures01::Future<Item = T, Error = RpcError> + Send + 'static>;
#[rpc]
pub trait Rpc {
#[rpc(name = "listDatabases")]
fn list_databases(&self, input: IntrospectionInput) -> RpcFutureResult<Vec<String>>;
#[rpc(name = "getDatabaseMetadata")]
fn get_database_metadata(&self, input: IntrospectionInput) -> RpcFutureResult<DatabaseMetadata>;
#[rpc(name = "getDatabaseDescription")]
fn get_database_description(&self, input: IntrospectionInput) -> RpcFutureResult<String>;
#[rpc(name = "introspect")]
fn introspect(&self, input: IntrospectionInput) -> RpcFutureResult<String>;
}
pub struct RpcImpl;
impl Rpc for RpcImpl {
fn list_databases(&self, input: IntrospectionInput) -> RpcFutureResult<Vec<String>> {
Box::new(Self::list_databases_internal(input.schema).boxed().compat())
}
fn get_database_metadata(&self, input: IntrospectionInput) -> RpcFutureResult<DatabaseMetadata> {
Box::new(Self::get_database_metadata_internal(input.schema).boxed().compat())
}
fn get_database_description(&self, input: IntrospectionInput) -> RpcFutureResult<String> {
Box::new(Self::get_database_description(input.schema).boxed().compat())
}
fn introspect(&self, input: IntrospectionInput) -> RpcFutureResult<String> {
Box::new(Self::introspect_internal(input.schema).boxed().compat())
}
}
impl RpcImpl {
pub fn new() -> Self {
RpcImpl
}
async fn load_connector(schema: &String) -> Result<Box<dyn IntrospectionConnector>, Error> {
let config = datamodel::parse_configuration(&schema)?;
let url = config
.datasources
.first()
.ok_or_else(|| CommandError::Generic(anyhow::anyhow!("There is no datasource in the schema.")))?
.url()
.to_owned()
.value;
Ok(Box::new(SqlIntrospectionConnector::new(&url).await?))
}
pub async fn introspect_internal(schema: String) -> RpcResult<String> {
let config = datamodel::parse_configuration(&schema).map_err(Error::from)?;
let url = config
.datasources
.first()
.ok_or_else(|| CommandError::Generic(anyhow::anyhow!("There is no datasource in the schema.")))
.map_err(Error::from)?
.url()
.to_owned()
.value;
let connector = RpcImpl::load_connector(&schema).await?;
let data_model = connector.introspect().await;
match data_model {
Ok(dm) if dm.models.is_empty() && dm.enums.is_empty() => Err(render_jsonrpc_error(Error::from(
CommandError::IntrospectionResultEmpty(url.to_string()),
))),
Ok(dm) => Ok(datamodel::render_datamodel_and_config_to_string(&dm, &config).map_err(Error::from)?),
Err(e) => Err(render_jsonrpc_error(Error::from(e))),
}
}
pub async fn list_databases_internal(schema: String) -> RpcResult<Vec<String>> {
let connector = RpcImpl::load_connector(&schema).await?;
Ok(connector.list_databases().await.map_err(Error::from)?)
}
pub async fn get_database_description(schema: String) -> RpcResult<String> {
let connector = RpcImpl::load_connector(&schema).await?;
Ok(connector.get_database_description().await.map_err(Error::from)?)
}
pub async fn get_database_metadata_internal(schema: String) -> RpcResult<DatabaseMetadata> {
let connector = RpcImpl::load_connector(&schema).await?;
Ok(connector.get_metadata().await.map_err(Error::from)?)
}
}
#[derive(Debug, Serialize, Deserialize)]
pub struct IntrospectionInput {
pub(crate) schema: String,
}