feat: add rqlite database support
This commit is contained in:
1 parent
a799958328
commit
0887b96236
34 files changed
+804
-12
No files matched your search
@@ -47,6 +47,7 @@ pub enum PoolKind {
|
||||
Mysql(db::mysql::MySqlPool, MysqlMode),
|
||||
Postgres(deadpool_postgres::Pool),
|
||||
Sqlite(db::sqlite::SqliteHandle),
|
||||
Rqlite(db::rqlite_driver::RqliteClient),
|
||||
Redis(db::redis_driver::RedisConnection),
|
||||
DuckDb(Arc<std::sync::Mutex<duckdb::Connection>>),
|
||||
MongoDb(mongodb::Client),
|
||||
@@ -343,6 +344,18 @@ impl AppState {
|
||||
db::sqlite::connect_path_with_extensions(&expand_tilde(&db_config.host), extensions).await?,
|
||||
)
|
||||
}
|
||||
DatabaseType::Rqlite => {
|
||||
let client = db::rqlite_driver::RqliteClient::new(
|
||||
&url,
|
||||
db_config.url_params.as_deref(),
|
||||
&db_config.username,
|
||||
&db_config.password,
|
||||
db_config.ssl,
|
||||
connect_timeout,
|
||||
)?;
|
||||
db::rqlite_driver::test_connection(&client, connect_timeout).await?;
|
||||
PoolKind::Rqlite(client)
|
||||
}
|
||||
DatabaseType::Redis => {
|
||||
let con = if db_config.uses_redis_cluster() {
|
||||
db::redis_driver::RedisConnection::Cluster(db::redis_driver::connect_cluster(&db_config).await?)
|
||||
@@ -919,6 +932,7 @@ pub async fn close_pool_kind(pool: PoolKind) {
|
||||
}
|
||||
PoolKind::Postgres(p) => p.close(),
|
||||
PoolKind::Sqlite(_) => {}
|
||||
PoolKind::Rqlite(_) => {}
|
||||
PoolKind::Redis(_) => {}
|
||||
PoolKind::DuckDb(con) => {
|
||||
crate::db::duckdb_driver::close_connection(con);
|
||||
|
||||
@@ -14,6 +14,7 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool {
|
||||
db_type,
|
||||
DatabaseType::Sqlite
|
||||
| DatabaseType::DuckDb
|
||||
| DatabaseType::Rqlite
|
||||
| DatabaseType::MongoDb
|
||||
| DatabaseType::Oracle
|
||||
| DatabaseType::Dameng
|
||||
|
||||
@@ -9,6 +9,7 @@ pub mod ob_oracle;
|
||||
pub mod postgres;
|
||||
pub mod proxy_tunnel;
|
||||
pub mod redis_driver;
|
||||
pub mod rqlite_driver;
|
||||
pub mod sqlite;
|
||||
pub mod sqlserver;
|
||||
pub mod ssh_tunnel;
|
||||
|
||||
@@ -0,0 +1,455 @@
|
||||
use reqwest::Client as HttpClient;
|
||||
use serde::Deserialize;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::with_connection_timeout;
|
||||
use crate::sql::starts_with_executable_sql_keyword;
|
||||
use crate::types::{
|
||||
ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectSource, ObjectSourceKind, QueryResult, TableInfo,
|
||||
TriggerInfo,
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct RqliteClient {
|
||||
http: HttpClient,
|
||||
base_url: String,
|
||||
query_params: String,
|
||||
auth: Option<(String, String)>,
|
||||
}
|
||||
|
||||
impl RqliteClient {
|
||||
pub fn new(
|
||||
url: &str,
|
||||
url_params: Option<&str>,
|
||||
username: &str,
|
||||
password: &str,
|
||||
tls_enabled: bool,
|
||||
timeout: Duration,
|
||||
) -> Result<Self, String> {
|
||||
let mut builder = HttpClient::builder().connect_timeout(timeout);
|
||||
if rqlite_accept_invalid_certs(tls_enabled, url_params) {
|
||||
builder = builder.danger_accept_invalid_certs(true);
|
||||
}
|
||||
if rqlite_should_bypass_system_proxy(url) {
|
||||
builder = builder.no_proxy();
|
||||
}
|
||||
let http = builder.build().map_err(|e| format!("Failed to configure rqlite HTTP client: {e}"))?;
|
||||
let auth = if username.trim().is_empty() { None } else { Some((username.to_string(), password.to_string())) };
|
||||
Ok(Self {
|
||||
http,
|
||||
base_url: url.trim_end_matches('/').split('?').next().unwrap_or(url).to_string(),
|
||||
query_params: normalize_rqlite_url_params(url_params),
|
||||
auth,
|
||||
})
|
||||
}
|
||||
|
||||
fn post_json(&self, path: &str, sql: &str) -> reqwest::RequestBuilder {
|
||||
let req = self.http.post(self.endpoint(path)).json(&[sql]);
|
||||
self.with_auth(req)
|
||||
}
|
||||
|
||||
fn with_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
||||
if let Some((ref user, ref pass)) = self.auth {
|
||||
req.basic_auth(user, Some(pass))
|
||||
} else {
|
||||
req
|
||||
}
|
||||
}
|
||||
|
||||
fn endpoint(&self, path: &str) -> String {
|
||||
if self.query_params.is_empty() {
|
||||
format!("{}{}", self.base_url, path)
|
||||
} else {
|
||||
format!("{}{}?{}", self.base_url, path, self.query_params)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct RqliteResponse {
|
||||
results: Vec<RqliteResult>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct RqliteResult {
|
||||
#[serde(default)]
|
||||
columns: Vec<String>,
|
||||
#[serde(default)]
|
||||
values: Vec<Vec<serde_json::Value>>,
|
||||
#[serde(default)]
|
||||
rows_affected: Option<u64>,
|
||||
#[serde(default)]
|
||||
error: Option<String>,
|
||||
}
|
||||
|
||||
enum RqliteEndpoint {
|
||||
Query,
|
||||
Execute,
|
||||
}
|
||||
|
||||
impl RqliteEndpoint {
|
||||
fn path(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Query => "/db/query",
|
||||
Self::Execute => "/db/execute",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn test_connection(client: &RqliteClient, timeout: Duration) -> Result<(), String> {
|
||||
with_connection_timeout("rqlite", timeout, async { query_one(client, "SELECT 1").await.map(|_| ()) }).await
|
||||
}
|
||||
|
||||
pub async fn list_databases(_client: &RqliteClient) -> Result<Vec<DatabaseInfo>, String> {
|
||||
Ok(vec![DatabaseInfo { name: "main".to_string() }])
|
||||
}
|
||||
|
||||
pub async fn list_tables(client: &RqliteClient, _schema: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let result = query_one(
|
||||
client,
|
||||
"SELECT name, type FROM sqlite_master WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name",
|
||||
)
|
||||
.await?;
|
||||
Ok(result
|
||||
.values
|
||||
.into_iter()
|
||||
.map(|row| {
|
||||
let table_type = value_as_string(row.get(1)).unwrap_or_else(|| "table".to_string());
|
||||
TableInfo {
|
||||
name: value_as_string(row.first()).unwrap_or_default(),
|
||||
table_type: if table_type.eq_ignore_ascii_case("view") { "VIEW" } else { "BASE TABLE" }.to_string(),
|
||||
comment: None,
|
||||
parent_schema: None,
|
||||
parent_name: None,
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(client: &RqliteClient, _schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
|
||||
let result = query_one(client, &format!("PRAGMA table_info({})", sqlite_ident(table))).await?;
|
||||
Ok(result
|
||||
.values
|
||||
.into_iter()
|
||||
.map(|row| ColumnInfo {
|
||||
name: value_by_column(&result.columns, &row, "name").unwrap_or_default(),
|
||||
data_type: value_by_column(&result.columns, &row, "type").unwrap_or_default(),
|
||||
is_nullable: value_by_column(&result.columns, &row, "notnull")
|
||||
.and_then(|value| value.parse::<i64>().ok())
|
||||
.unwrap_or(0)
|
||||
== 0,
|
||||
column_default: value_by_column(&result.columns, &row, "dflt_value"),
|
||||
is_primary_key: value_by_column(&result.columns, &row, "pk")
|
||||
.and_then(|value| value.parse::<i64>().ok())
|
||||
.unwrap_or(0)
|
||||
> 0,
|
||||
extra: None,
|
||||
comment: None,
|
||||
numeric_precision: None,
|
||||
numeric_scale: None,
|
||||
character_maximum_length: None,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_indexes(client: &RqliteClient, _schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
let result = query_one(client, &format!("PRAGMA index_list({})", sqlite_ident(table))).await?;
|
||||
let mut indexes = Vec::new();
|
||||
|
||||
for row in result.values {
|
||||
let name = value_by_column(&result.columns, &row, "name").unwrap_or_default();
|
||||
if name.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let is_unique =
|
||||
value_by_column(&result.columns, &row, "unique").and_then(|value| value.parse::<i64>().ok()).unwrap_or(0)
|
||||
!= 0;
|
||||
let origin = value_by_column(&result.columns, &row, "origin").unwrap_or_default();
|
||||
let column_result = query_one(client, &format!("PRAGMA index_info({})", sqlite_ident(&name))).await?;
|
||||
let columns = column_result
|
||||
.values
|
||||
.iter()
|
||||
.filter_map(|row| value_by_column(&column_result.columns, row, "name"))
|
||||
.collect();
|
||||
indexes.push(IndexInfo {
|
||||
name,
|
||||
columns,
|
||||
is_unique,
|
||||
is_primary: origin == "pk",
|
||||
filter: None,
|
||||
index_type: None,
|
||||
included_columns: None,
|
||||
comment: None,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(indexes)
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(
|
||||
client: &RqliteClient,
|
||||
_schema: &str,
|
||||
table: &str,
|
||||
) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
let result = query_one(client, &format!("PRAGMA foreign_key_list({})", sqlite_ident(table))).await?;
|
||||
Ok(result
|
||||
.values
|
||||
.into_iter()
|
||||
.map(|row| ForeignKeyInfo {
|
||||
name: format!("fk_{}", value_by_column(&result.columns, &row, "id").unwrap_or_else(|| "0".to_string())),
|
||||
column: value_by_column(&result.columns, &row, "from").unwrap_or_default(),
|
||||
ref_schema: None,
|
||||
ref_table: value_by_column(&result.columns, &row, "table").unwrap_or_default(),
|
||||
ref_column: value_by_column(&result.columns, &row, "to").unwrap_or_default(),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(client: &RqliteClient, _schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let result = query_one(
|
||||
client,
|
||||
&format!(
|
||||
"SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = {} ORDER BY name",
|
||||
sqlite_string(table)
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
Ok(result
|
||||
.values
|
||||
.into_iter()
|
||||
.map(|row| {
|
||||
let sql_text = value_as_string(row.get(1)).unwrap_or_default().to_uppercase();
|
||||
let timing = if sql_text.contains("BEFORE") {
|
||||
"BEFORE"
|
||||
} else if sql_text.contains("AFTER") {
|
||||
"AFTER"
|
||||
} else {
|
||||
"INSTEAD OF"
|
||||
};
|
||||
let event = if sql_text.contains("INSERT") {
|
||||
"INSERT"
|
||||
} else if sql_text.contains("UPDATE") {
|
||||
"UPDATE"
|
||||
} else {
|
||||
"DELETE"
|
||||
};
|
||||
TriggerInfo {
|
||||
name: value_as_string(row.first()).unwrap_or_default(),
|
||||
event: event.to_string(),
|
||||
timing: timing.to_string(),
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn table_ddl(client: &RqliteClient, table: &str) -> Result<String, String> {
|
||||
first_string_cell(
|
||||
query_one(
|
||||
client,
|
||||
&format!("SELECT sql FROM sqlite_master WHERE type='table' AND name={}", sqlite_string(table)),
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn object_source(
|
||||
client: &RqliteClient,
|
||||
name: &str,
|
||||
object_type: &ObjectSourceKind,
|
||||
) -> Result<ObjectSource, String> {
|
||||
let kind = match object_type {
|
||||
ObjectSourceKind::View => "view",
|
||||
_ => return Err("Object source is not supported for this rqlite object type".to_string()),
|
||||
};
|
||||
let source = first_string_cell(
|
||||
query_one(
|
||||
client,
|
||||
&format!(
|
||||
"SELECT sql FROM sqlite_master WHERE type={} AND name={}",
|
||||
sqlite_string(kind),
|
||||
sqlite_string(name)
|
||||
),
|
||||
)
|
||||
.await?,
|
||||
)?;
|
||||
Ok(ObjectSource { name: name.to_string(), object_type: object_type.clone(), schema: None, source })
|
||||
}
|
||||
|
||||
pub async fn execute_query(client: &RqliteClient, sql: &str) -> Result<QueryResult, String> {
|
||||
execute_query_with_max_rows(client, sql, None).await
|
||||
}
|
||||
|
||||
pub async fn execute_query_with_max_rows(
|
||||
client: &RqliteClient,
|
||||
sql: &str,
|
||||
max_rows: Option<usize>,
|
||||
) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
if starts_with_executable_sql_keyword(sql, &["SELECT", "PRAGMA", "EXPLAIN", "WITH"]) {
|
||||
let result = query_one(client, sql).await?;
|
||||
Ok(query_result_from_rqlite_result(result, start.elapsed().as_millis(), max_rows))
|
||||
} else {
|
||||
let result = execute_one(client, sql).await?;
|
||||
let affected_rows = result.rows_affected.unwrap_or(0);
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
session_id: None,
|
||||
has_more: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async fn query_one(client: &RqliteClient, sql: &str) -> Result<RqliteResult, String> {
|
||||
request_one(client, RqliteEndpoint::Query, sql).await
|
||||
}
|
||||
|
||||
async fn execute_one(client: &RqliteClient, sql: &str) -> Result<RqliteResult, String> {
|
||||
request_one(client, RqliteEndpoint::Execute, sql).await
|
||||
}
|
||||
|
||||
async fn request_one(client: &RqliteClient, endpoint: RqliteEndpoint, sql: &str) -> Result<RqliteResult, String> {
|
||||
let resp =
|
||||
client.post_json(endpoint.path(), sql).send().await.map_err(|e| format!("rqlite request failed: {e}"))?;
|
||||
let status = resp.status();
|
||||
let body = resp.text().await.map_err(|e| format!("rqlite response read failed: {e}"))?;
|
||||
if !status.is_success() {
|
||||
return Err(format!("rqlite error ({status}): {body}"));
|
||||
}
|
||||
let response: RqliteResponse =
|
||||
serde_json::from_str(&body).map_err(|e| format!("rqlite parse error: {e}; body: {body}"))?;
|
||||
let result = response.results.into_iter().next().ok_or_else(|| "rqlite returned no result".to_string())?;
|
||||
if let Some(error) = result.error.as_ref().filter(|error| !error.is_empty()) {
|
||||
return Err(format!("rqlite error: {error}"));
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
fn query_result_from_rqlite_result(
|
||||
mut result: RqliteResult,
|
||||
execution_time_ms: u128,
|
||||
max_rows: Option<usize>,
|
||||
) -> QueryResult {
|
||||
let row_limit = max_rows.unwrap_or(crate::query::MAX_ROWS).max(1);
|
||||
let truncated = result.values.len() > row_limit;
|
||||
if truncated {
|
||||
result.values.truncate(row_limit);
|
||||
}
|
||||
QueryResult {
|
||||
columns: result.columns,
|
||||
rows: result.values,
|
||||
affected_rows: 0,
|
||||
execution_time_ms,
|
||||
truncated,
|
||||
session_id: None,
|
||||
has_more: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn first_string_cell(result: RqliteResult) -> Result<String, String> {
|
||||
result
|
||||
.values
|
||||
.first()
|
||||
.and_then(|row| row.first())
|
||||
.and_then(|value| value_as_string(Some(value)))
|
||||
.ok_or_else(|| "Object not found".to_string())
|
||||
}
|
||||
|
||||
fn value_by_column(columns: &[String], row: &[serde_json::Value], name: &str) -> Option<String> {
|
||||
columns
|
||||
.iter()
|
||||
.position(|column| column.eq_ignore_ascii_case(name))
|
||||
.and_then(|index| row.get(index))
|
||||
.and_then(|value| value_as_string(Some(value)))
|
||||
}
|
||||
|
||||
fn value_as_string(value: Option<&serde_json::Value>) -> Option<String> {
|
||||
match value? {
|
||||
serde_json::Value::Null => None,
|
||||
serde_json::Value::String(value) => Some(value.clone()),
|
||||
serde_json::Value::Number(value) => Some(value.to_string()),
|
||||
serde_json::Value::Bool(value) => Some(value.to_string()),
|
||||
other => Some(other.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn sqlite_ident(value: &str) -> String {
|
||||
format!("\"{}\"", value.replace('"', "\"\""))
|
||||
}
|
||||
|
||||
fn sqlite_string(value: &str) -> String {
|
||||
format!("'{}'", value.replace('\'', "''"))
|
||||
}
|
||||
|
||||
fn normalize_rqlite_url_params(params: Option<&str>) -> String {
|
||||
params
|
||||
.unwrap_or("")
|
||||
.trim()
|
||||
.trim_start_matches('?')
|
||||
.split('&')
|
||||
.filter(|part| !part.is_empty())
|
||||
.collect::<Vec<_>>()
|
||||
.join("&")
|
||||
}
|
||||
|
||||
fn rqlite_accept_invalid_certs(tls_enabled: bool, url_params: Option<&str>) -> bool {
|
||||
tls_enabled
|
||||
&& url_params
|
||||
.unwrap_or("")
|
||||
.trim()
|
||||
.trim_start_matches('?')
|
||||
.split('&')
|
||||
.filter_map(|pair| pair.split_once('='))
|
||||
.any(|(key, value)| {
|
||||
matches!(key.trim().to_ascii_lowercase().as_str(), "insecure" | "tls_insecure" | "accept_invalid_certs")
|
||||
&& matches!(value.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes" | "on")
|
||||
})
|
||||
}
|
||||
|
||||
fn rqlite_should_bypass_system_proxy(base_url: &str) -> bool {
|
||||
let Ok(parsed) = reqwest::Url::parse(base_url) else {
|
||||
return false;
|
||||
};
|
||||
let Some(host) = parsed.host_str() else {
|
||||
return false;
|
||||
};
|
||||
let host = host.trim_matches(['[', ']']);
|
||||
host.eq_ignore_ascii_case("localhost") || host.parse::<std::net::IpAddr>().is_ok_and(|ip| ip.is_loopback())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn converts_query_result_and_truncates_probe_rows() {
|
||||
let result = RqliteResult {
|
||||
columns: vec!["id".to_string(), "name".to_string()],
|
||||
values: vec![
|
||||
vec![serde_json::json!(1), serde_json::json!("Ada")],
|
||||
vec![serde_json::json!(2), serde_json::json!("Linus")],
|
||||
],
|
||||
rows_affected: None,
|
||||
error: None,
|
||||
};
|
||||
|
||||
let result = query_result_from_rqlite_result(result, 8, Some(1));
|
||||
|
||||
assert_eq!(result.columns, vec!["id", "name"]);
|
||||
assert_eq!(result.rows, vec![vec![serde_json::json!(1), serde_json::json!("Ada")]]);
|
||||
assert_eq!(result.execution_time_ms, 8);
|
||||
assert!(result.truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn maps_column_values_by_name() {
|
||||
let columns = vec!["name".to_string(), "notnull".to_string(), "pk".to_string()];
|
||||
let row = vec![serde_json::json!("id"), serde_json::json!(1), serde_json::json!(1)];
|
||||
|
||||
assert_eq!(value_by_column(&columns, &row, "NAME").as_deref(), Some("id"));
|
||||
assert_eq!(value_by_column(&columns, &row, "notnull").as_deref(), Some("1"));
|
||||
}
|
||||
}
|
||||
@@ -169,6 +169,7 @@ pub enum DatabaseType {
|
||||
Mysql,
|
||||
Postgres,
|
||||
Sqlite,
|
||||
Rqlite,
|
||||
Redis,
|
||||
#[serde(rename = "duckdb")]
|
||||
DuckDb,
|
||||
@@ -301,6 +302,7 @@ impl ConnectionConfig {
|
||||
},
|
||||
DatabaseType::Redshift => Some("dev"),
|
||||
DatabaseType::ClickHouse => Some("default"),
|
||||
DatabaseType::Rqlite => Some("main"),
|
||||
DatabaseType::Gaussdb | DatabaseType::OpenGauss => Some("postgres"),
|
||||
DatabaseType::Kingbase | DatabaseType::Vastbase => Some("postgres"),
|
||||
DatabaseType::Highgo => Some("highgo"),
|
||||
@@ -386,6 +388,7 @@ impl ConnectionConfig {
|
||||
format!("postgres://{host}:{port}{db_part}{suffix}")
|
||||
}
|
||||
DatabaseType::ClickHouse => clickhouse_http_url(self, raw_host, port),
|
||||
DatabaseType::Rqlite => rqlite_http_url(self, raw_host, port),
|
||||
DatabaseType::SqlServer => {
|
||||
format!("server=tcp:{host},{port};database={}", self.database.as_deref().unwrap_or("master"))
|
||||
}
|
||||
@@ -489,6 +492,7 @@ impl ConnectionConfig {
|
||||
format!("postgres://{}:{}@{host}:{port}{db_part}{suffix}", username, password)
|
||||
}
|
||||
DatabaseType::ClickHouse => clickhouse_http_url(self, raw_host, port),
|
||||
DatabaseType::Rqlite => rqlite_http_url(self, raw_host, port),
|
||||
DatabaseType::SqlServer => format!(
|
||||
"server=tcp:{host},{port};user={};password={};database={}",
|
||||
self.username,
|
||||
@@ -825,6 +829,31 @@ fn clickhouse_http_url(config: &ConnectionConfig, host: &str, port: u16) -> Stri
|
||||
format!("{scheme}://{}:{port}", bracket_ipv6(trimmed))
|
||||
}
|
||||
|
||||
fn rqlite_http_url(config: &ConnectionConfig, host: &str, port: u16) -> String {
|
||||
let trimmed = host.trim();
|
||||
if let Some(rest) = trimmed.strip_prefix("https://") {
|
||||
return format!("https://{}", trim_http_host_port(rest, port));
|
||||
}
|
||||
if let Some(rest) = trimmed.strip_prefix("http://") {
|
||||
let scheme = if config.ssl { "https" } else { "http" };
|
||||
return format!("{scheme}://{}", trim_http_host_port(rest, port));
|
||||
}
|
||||
let scheme = if config.ssl { "https" } else { "http" };
|
||||
format!("{scheme}://{}:{port}", bracket_ipv6(trimmed))
|
||||
}
|
||||
|
||||
fn trim_http_host_port(value: &str, default_port: u16) -> String {
|
||||
let authority = value.trim_end_matches('/').split('/').next().unwrap_or(value).split('?').next().unwrap_or(value);
|
||||
if authority.starts_with('[') && !authority.contains("]:") {
|
||||
return format!("{authority}:{default_port}");
|
||||
}
|
||||
if authority.rsplit_once(':').is_some() {
|
||||
authority.to_string()
|
||||
} else {
|
||||
format!("{authority}:{default_port}")
|
||||
}
|
||||
}
|
||||
|
||||
fn trim_clickhouse_host_port(value: &str, default_port: u16) -> String {
|
||||
let authority = value.trim_end_matches('/').split('/').next().unwrap_or(value).split('?').next().unwrap_or(value);
|
||||
if authority.starts_with('[') && !authority.contains("]:") {
|
||||
|
||||
@@ -564,6 +564,17 @@ pub async fn do_execute(
|
||||
wait_for_query_opt(cancel_token, query_timeout, db::sqlite::execute_query_with_max_rows(&p, sql, max_rows))
|
||||
.await
|
||||
}
|
||||
PoolKind::Rqlite(client) => {
|
||||
let client = client.clone();
|
||||
let max_rows = options.max_rows;
|
||||
drop(connections);
|
||||
wait_for_query_opt(
|
||||
cancel_token,
|
||||
query_timeout,
|
||||
db::rqlite_driver::execute_query_with_max_rows(&client, sql, max_rows),
|
||||
)
|
||||
.await
|
||||
}
|
||||
PoolKind::ClickHouse(client) => {
|
||||
let client = client.clone();
|
||||
let database = pool_key.split(':').nth(1).unwrap_or("default").to_string();
|
||||
@@ -1107,7 +1118,9 @@ pub async fn execute_statements_in_transaction(
|
||||
PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()),
|
||||
PoolKind::Mysql(mp, _mode) => TxPath::Mysql(mp.clone(), false),
|
||||
PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()),
|
||||
PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Agent(_) => TxPath::Explicit,
|
||||
PoolKind::ClickHouse(_) | PoolKind::Rqlite(_) | PoolKind::SqlServer(_) | PoolKind::Agent(_) => {
|
||||
TxPath::Explicit
|
||||
}
|
||||
PoolKind::DuckDb(_)
|
||||
| PoolKind::Redis(_)
|
||||
| PoolKind::MongoDb(_)
|
||||
|
||||
@@ -300,6 +300,7 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Ve
|
||||
PoolKind::Mysql(p, mode) => dispatch_mysql!(p, mode, db::mysql::list_databases, db::ob_oracle::list_databases),
|
||||
PoolKind::Postgres(p) => db::postgres::list_databases(p).await,
|
||||
PoolKind::Sqlite(p) => db::sqlite::list_databases(p).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::list_databases(client).await,
|
||||
PoolKind::DuckDb(con) => {
|
||||
let con = con.lock().map_err(|e| e.to_string())?;
|
||||
duckdb_list_databases_with_attached(&con, &duckdb_attached_names)
|
||||
@@ -428,6 +429,9 @@ async fn list_tables_once(
|
||||
PoolKind::Sqlite(p) => {
|
||||
db::sqlite::list_tables(p, schema).await.map(|tables| filter_table_infos(tables, filter, limit))
|
||||
}
|
||||
PoolKind::Rqlite(client) => {
|
||||
db::rqlite_driver::list_tables(client, schema).await.map(|tables| filter_table_infos(tables, filter, limit))
|
||||
}
|
||||
PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database)
|
||||
.await
|
||||
.map(|names| collection_names_to_tables(names, "COLLECTION"))
|
||||
@@ -825,6 +829,9 @@ pub async fn get_columns_core(
|
||||
}
|
||||
PoolKind::Postgres(p) => db::postgres::get_columns(p, schema, table).await.map(deduplicate_column_infos),
|
||||
PoolKind::Sqlite(p) => db::sqlite::get_columns(p, schema, table).await.map(deduplicate_column_infos),
|
||||
PoolKind::Rqlite(client) => {
|
||||
db::rqlite_driver::get_columns(client, schema, table).await.map(deduplicate_column_infos)
|
||||
}
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
}
|
||||
@@ -896,6 +903,7 @@ pub async fn list_indexes_core(
|
||||
}
|
||||
PoolKind::Postgres(p) => db::postgres::list_indexes(p, schema, table).await,
|
||||
PoolKind::Sqlite(p) => db::sqlite::list_indexes(p, schema, table).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::list_indexes(client, schema, table).await,
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
}
|
||||
@@ -924,6 +932,7 @@ pub async fn list_foreign_keys_core(
|
||||
}
|
||||
PoolKind::Postgres(p) => db::postgres::list_foreign_keys(p, schema, table).await,
|
||||
PoolKind::Sqlite(p) => db::sqlite::list_foreign_keys(p, schema, table).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::list_foreign_keys(client, schema, table).await,
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
}
|
||||
@@ -952,6 +961,7 @@ pub async fn list_triggers_core(
|
||||
}
|
||||
PoolKind::Postgres(p) => db::postgres::list_triggers(p, schema, table).await,
|
||||
PoolKind::Sqlite(p) => db::sqlite::list_triggers(p, schema, table).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::list_triggers(client, schema, table).await,
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
}
|
||||
@@ -1019,6 +1029,7 @@ pub async fn get_table_ddl_core(
|
||||
}
|
||||
PoolKind::Postgres(p) => pg_ddl(p, schema, table).await,
|
||||
PoolKind::Sqlite(p) => sqlite_ddl(p, table).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::table_ddl(client, table).await,
|
||||
_ => Err("DDL not supported for this database type".to_string()),
|
||||
}
|
||||
}
|
||||
@@ -1244,6 +1255,9 @@ pub async fn get_object_source_core(
|
||||
PoolKind::Sqlite(pool) => first_string_cell(
|
||||
db::sqlite::execute_query(pool, &sqlite_object_source_sql(name, &object_type)).await?,
|
||||
)?,
|
||||
PoolKind::Rqlite(client) => {
|
||||
return db::rqlite_driver::object_source(client, name, &object_type).await;
|
||||
}
|
||||
PoolKind::ClickHouse(client) if matches!(object_type, db::ObjectSourceKind::View) => {
|
||||
let result = db::clickhouse_driver::execute_query(
|
||||
client,
|
||||
|
||||
@@ -241,7 +241,7 @@ fn capabilities_for(database_type: Option<DatabaseType>) -> TableStructureCapabi
|
||||
comment: true,
|
||||
..base
|
||||
},
|
||||
Some(DatabaseType::Sqlite) => TableStructureCapabilities {
|
||||
Some(DatabaseType::Sqlite | DatabaseType::Rqlite) => TableStructureCapabilities {
|
||||
dialect: StructureDialect::Sqlite,
|
||||
add_column: true,
|
||||
drop_column: true,
|
||||
@@ -1966,6 +1966,34 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_rqlite_changes_with_sqlite_dialect() {
|
||||
let mut email = column("email");
|
||||
email.data_type = "text".to_string();
|
||||
email.is_nullable = false;
|
||||
let mut email_index = index("idx_users_email", &["email"]);
|
||||
email_index.filter = "email IS NOT NULL".to_string();
|
||||
|
||||
let result = build_table_structure_change_sql(TableStructureSqlOptions {
|
||||
database_type: Some(DatabaseType::Rqlite),
|
||||
schema: None,
|
||||
table_name: "users".to_string(),
|
||||
columns: vec![email],
|
||||
indexes: vec![email_index],
|
||||
table_comment: None,
|
||||
original_table_comment: None,
|
||||
});
|
||||
|
||||
assert_eq!(result.warnings, Vec::<String>::new());
|
||||
assert_eq!(
|
||||
result.statements,
|
||||
vec![
|
||||
"ALTER TABLE \"users\" ADD COLUMN \"email\" text NOT NULL;",
|
||||
"CREATE INDEX \"idx_users_email\" ON \"users\" (\"email\") WHERE email IS NOT NULL;",
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_mysql_column_reorder_statements() {
|
||||
let mut id = column("id");
|
||||
|
||||
Reference in new issue
Block a user