mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 06:00:39 +00:00
fix(mongodb): support shell projection and db.version
This commit is contained in:
@@ -39,6 +39,7 @@ public final class AgentProtocol {
|
||||
public static final String MONGO_METHOD_LIST_DATABASES = "list_databases";
|
||||
public static final String MONGO_METHOD_LIST_COLLECTIONS = "list_collections";
|
||||
public static final String MONGO_METHOD_FIND_DOCUMENTS = "find_documents";
|
||||
public static final String MONGO_METHOD_SERVER_VERSION = "server_version";
|
||||
public static final String MONGO_METHOD_INSERT_DOCUMENT = "insert_document";
|
||||
public static final String MONGO_METHOD_UPDATE_DOCUMENT = "update_document";
|
||||
public static final String MONGO_METHOD_DELETE_DOCUMENT = "delete_document";
|
||||
@@ -113,6 +114,7 @@ public final class AgentProtocol {
|
||||
MONGO_METHOD_LIST_DATABASES,
|
||||
MONGO_METHOD_LIST_COLLECTIONS,
|
||||
MONGO_METHOD_FIND_DOCUMENTS,
|
||||
MONGO_METHOD_SERVER_VERSION,
|
||||
MONGO_METHOD_INSERT_DOCUMENT,
|
||||
MONGO_METHOD_UPDATE_DOCUMENT,
|
||||
MONGO_METHOD_DELETE_DOCUMENT
|
||||
|
||||
@@ -68,6 +68,7 @@
|
||||
"list_databases",
|
||||
"list_collections",
|
||||
"find_documents",
|
||||
"server_version",
|
||||
"insert_document",
|
||||
"update_document",
|
||||
"delete_document"
|
||||
|
||||
@@ -61,6 +61,11 @@ public final class MongoAgent {
|
||||
return element == null || element instanceof JsonNull ? null : element.getAsString();
|
||||
}
|
||||
|
||||
static Document documentOrNull(JsonObject object, String key) {
|
||||
String json = stringOrNull(object, key);
|
||||
return json == null || json.isBlank() ? null : Document.parse(json);
|
||||
}
|
||||
|
||||
static MongoClientSettings.Builder configureBuilder(JsonObject connObj) {
|
||||
String host = connObj.has("host") ? connObj.get("host").getAsString() : "127.0.0.1";
|
||||
int port = connObj.has("port") ? connObj.get("port").getAsInt() : 27017;
|
||||
@@ -350,16 +355,22 @@ public final class MongoAgent {
|
||||
String collection = params.get("collection").getAsString();
|
||||
long skip = params.has("skip") ? params.get("skip").getAsLong() : 0;
|
||||
int limit = params.has("limit") ? params.get("limit").getAsInt() : 50;
|
||||
String filterJson = stringOrNull(params, "filter");
|
||||
String sortJson = stringOrNull(params, "sort");
|
||||
Document filterDoc = documentOrNull(params, "filter");
|
||||
Document projectionDoc = documentOrNull(params, "projection");
|
||||
Document sortDoc = documentOrNull(params, "sort");
|
||||
|
||||
var col = c.getDatabase(database).getCollection(collection);
|
||||
Document filterDoc = filterJson != null && !filterJson.isBlank() ? Document.parse(filterJson) : new Document();
|
||||
if (filterDoc == null) {
|
||||
filterDoc = new Document();
|
||||
}
|
||||
long total = col.countDocuments(filterDoc);
|
||||
|
||||
var iterable = col.find(filterDoc).skip((int) skip).limit(limit);
|
||||
if (sortJson != null && !sortJson.isBlank()) {
|
||||
iterable = iterable.sort(Document.parse(sortJson));
|
||||
if (projectionDoc != null) {
|
||||
iterable = iterable.projection(projectionDoc);
|
||||
}
|
||||
if (sortDoc != null) {
|
||||
iterable = iterable.sort(sortDoc);
|
||||
}
|
||||
|
||||
List<Map<String, Object>> documents = new ArrayList<>();
|
||||
@@ -372,6 +383,21 @@ public final class MongoAgent {
|
||||
return result;
|
||||
}
|
||||
|
||||
private static Object serverVersion(JsonObject params) {
|
||||
MongoClient c = requireClient();
|
||||
String database = defaultString(stringOrNull(params, "database"), "admin");
|
||||
Document buildInfo = c.getDatabase(database).runCommand(new Document("buildInfo", 1));
|
||||
return serverVersionFromBuildInfo(buildInfo);
|
||||
}
|
||||
|
||||
static String serverVersionFromBuildInfo(Document buildInfo) {
|
||||
String version = buildInfo.getString("version");
|
||||
if (version == null || version.isBlank()) {
|
||||
throw new IllegalStateException("MongoDB server version not found");
|
||||
}
|
||||
return version;
|
||||
}
|
||||
|
||||
private static Object insertDocument(JsonObject params) {
|
||||
MongoClient c = requireClient();
|
||||
String database = params.get("database").getAsString();
|
||||
@@ -559,6 +585,7 @@ public final class MongoAgent {
|
||||
case AgentProtocol.MONGO_METHOD_LIST_COLLECTIONS -> listCollections(params);
|
||||
case AgentProtocol.METHOD_LIST_INDEXES -> listIndexes(params);
|
||||
case AgentProtocol.MONGO_METHOD_FIND_DOCUMENTS -> findDocuments(params);
|
||||
case AgentProtocol.MONGO_METHOD_SERVER_VERSION -> serverVersion(params);
|
||||
case AgentProtocol.MONGO_METHOD_INSERT_DOCUMENT -> insertDocument(params);
|
||||
case AgentProtocol.MONGO_METHOD_UPDATE_DOCUMENT -> updateDocument(params);
|
||||
case AgentProtocol.MONGO_METHOD_DELETE_DOCUMENT -> deleteDocument(params);
|
||||
|
||||
@@ -108,6 +108,39 @@ class MongoAgentTest {
|
||||
assertFalse(json.getAsJsonObject("error").get("message").getAsString().contains("Unknown method"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesOptionalDocumentParameters() {
|
||||
JsonObject params = new JsonObject();
|
||||
params.addProperty("projection", "{\"title\":1,\"_id\":0}");
|
||||
params.addProperty("filter", "");
|
||||
|
||||
Document projection = MongoAgent.documentOrNull(params, "projection");
|
||||
|
||||
assertNotNull(projection);
|
||||
assertEquals(1, projection.get("title"));
|
||||
assertEquals(0, projection.get("_id"));
|
||||
assertEquals(null, MongoAgent.documentOrNull(params, "filter"));
|
||||
assertEquals(null, MongoAgent.documentOrNull(params, "sort"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void serverVersionMethodIsRecognizedOverJsonRpc() {
|
||||
String response = MongoAgent.handleRequest(
|
||||
"{\"jsonrpc\":\"2.0\",\"id\":9,\"method\":\"server_version\","
|
||||
+ "\"params\":{\"database\":\"admin\"}}");
|
||||
|
||||
JsonObject json = JsonParser.parseString(response).getAsJsonObject();
|
||||
assertEquals(9, json.get("id").getAsInt());
|
||||
assertEquals("Not connected", json.getAsJsonObject("error").get("message").getAsString());
|
||||
assertFalse(json.getAsJsonObject("error").get("message").getAsString().contains("Unknown method"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void extractsServerVersionFromBuildInfo() {
|
||||
assertEquals("4.4.29", MongoAgent.serverVersionFromBuildInfo(new Document("version", "4.4.29")));
|
||||
assertThrows(IllegalStateException.class, () -> MongoAgent.serverVersionFromBuildInfo(new Document("ok", 1)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void convertsMongoIndexDocumentToIndexInfo() {
|
||||
Document index = new Document("name", "idx_user_status")
|
||||
|
||||
@@ -396,7 +396,7 @@ async function load() {
|
||||
try {
|
||||
const filter = currentDocumentFilter();
|
||||
const sort = sortInput.value.trim() || undefined;
|
||||
const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, page.value * pageSize.value, pageSize.value, filter, sort, executionId);
|
||||
const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, page.value * pageSize.value, pageSize.value, filter, undefined, sort, executionId);
|
||||
if (documentLoadExecutionId.value !== executionId) return;
|
||||
const nextDocuments = result.documents.map(asRecord);
|
||||
documents.value = nextDocuments;
|
||||
|
||||
@@ -369,6 +369,7 @@ export const mongoDropDatabase = forward("mongoDropDatabase");
|
||||
export const mongoDropCollection = forward("mongoDropCollection");
|
||||
export const documentFindDocuments = forward("documentFindDocuments");
|
||||
export const mongoFindDocuments = forward("mongoFindDocuments");
|
||||
export const mongoServerVersion = forward("mongoServerVersion");
|
||||
export const mongoAggregateDocuments = forward("mongoAggregateDocuments");
|
||||
export const mongoInsertDocument = forward("mongoInsertDocument");
|
||||
export const mongoInsertDocuments = forward("mongoInsertDocuments");
|
||||
|
||||
@@ -1720,12 +1720,16 @@ export async function vectorListCollections(connectionId: string): Promise<Colle
|
||||
return mongoListCollections(connectionId, "default");
|
||||
}
|
||||
|
||||
export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, sort, executionId });
|
||||
export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return post("/api/document-store/find-documents", { connectionId, database, collection, skip, limit, filter, sort, executionId });
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return post("/api/document-store/find-documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function mongoServerVersion(connectionId: string, database: string, executionId?: string): Promise<string> {
|
||||
return post("/api/mongo/server-version", { connectionId, database, executionId });
|
||||
}
|
||||
|
||||
export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number, executionId?: string): Promise<MongoDocumentResult> {
|
||||
|
||||
@@ -3,6 +3,7 @@ import type { QueryResult } from "@/types/database";
|
||||
export interface MongoFindCommand {
|
||||
collection: string;
|
||||
filter: string;
|
||||
projection?: string;
|
||||
skip: number;
|
||||
limit: number;
|
||||
sort?: string;
|
||||
@@ -26,6 +27,10 @@ export interface MongoUseCommand {
|
||||
database: string;
|
||||
}
|
||||
|
||||
export interface MongoVersionCommand {
|
||||
kind: "version";
|
||||
}
|
||||
|
||||
export type MongoWriteCommand = { kind: "insert"; collection: string; docsJson: string } | { kind: "update"; collection: string; filter: string; update: string; many: boolean } | { kind: "delete"; collection: string; filter: string; many: boolean };
|
||||
|
||||
export interface MongoAggregateSafetyOptions {
|
||||
@@ -45,8 +50,15 @@ export function parseMongoFindCommand(input: string): MongoFindCommand | null {
|
||||
if (findCloseIndex < 0) return null;
|
||||
|
||||
const findArgs = splitTopLevel(source.slice(findOpenIndex + 1, findCloseIndex));
|
||||
if (findArgs.length > 2 && findArgs.slice(2).some((arg) => arg.trim())) return null;
|
||||
const filter = normalizeJsonArgument(findArgs[0] || "{}");
|
||||
if (!filter) return null;
|
||||
let projection: string | undefined;
|
||||
if (findArgs[1]?.trim()) {
|
||||
const parsedProjection = normalizeJsonArgument(findArgs[1]);
|
||||
if (!parsedProjection) return null;
|
||||
projection = parsedProjection;
|
||||
}
|
||||
|
||||
const chain = source.slice(findCloseIndex + 1).trim();
|
||||
if (chain && !chain.startsWith(".")) return null;
|
||||
@@ -66,6 +78,7 @@ export function parseMongoFindCommand(input: string): MongoFindCommand | null {
|
||||
return {
|
||||
collection: target.collection,
|
||||
filter,
|
||||
...(projection ? { projection } : {}),
|
||||
skip,
|
||||
limit,
|
||||
sort,
|
||||
@@ -143,6 +156,11 @@ export function parseMongoUseCommand(input: string): MongoUseCommand | null {
|
||||
};
|
||||
}
|
||||
|
||||
export function parseMongoVersionCommand(input: string): MongoVersionCommand | null {
|
||||
const source = input.trim().replace(/;$/, "").trim();
|
||||
return /^db\s*\.\s*version\s*\(\s*\)$/i.test(source) ? { kind: "version" } : null;
|
||||
}
|
||||
|
||||
export function parseMongoWriteCommand(input: string): MongoWriteCommand | null {
|
||||
const source = input.trim().replace(/;$/, "").trim();
|
||||
const insertOne = parseCollectionMethodTarget(source, "insertOne");
|
||||
@@ -273,6 +291,15 @@ export function mongoUseToQueryResult(database: string, executionTimeMs: number)
|
||||
};
|
||||
}
|
||||
|
||||
export function mongoVersionToQueryResult(version: string, executionTimeMs: number): QueryResult {
|
||||
return {
|
||||
columns: ["version"],
|
||||
rows: [[version]],
|
||||
affected_rows: 1,
|
||||
execution_time_ms: Math.max(0, Math.round(executionTimeMs)),
|
||||
};
|
||||
}
|
||||
|
||||
export function mongoIndexesToQueryResult(
|
||||
indexes: {
|
||||
name: string;
|
||||
|
||||
@@ -1450,12 +1450,16 @@ export async function vectorListCollections(connectionId: string): Promise<Colle
|
||||
return mongoListCollections(connectionId, "default");
|
||||
}
|
||||
|
||||
export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, sort, executionId });
|
||||
export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return invoke("document_find_documents", { connectionId, database, collection, skip, limit, filter, sort, executionId });
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
return invoke("document_find_documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function mongoServerVersion(connectionId: string, database: string, executionId?: string): Promise<string> {
|
||||
return invoke("mongo_server_version", { connectionId, database, executionId });
|
||||
}
|
||||
|
||||
export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number, executionId?: string): Promise<MongoDocumentResult> {
|
||||
|
||||
@@ -14,14 +14,16 @@ import {
|
||||
mongoCountToQueryResult,
|
||||
mongoDocumentsToQueryResult,
|
||||
mongoIndexesToQueryResult,
|
||||
mongoWriteToQueryResult,
|
||||
mongoUseToQueryResult,
|
||||
mongoVersionToQueryResult,
|
||||
mongoWriteToQueryResult,
|
||||
parseMongoAggregateCommand,
|
||||
parseMongoCountDocumentsCommand,
|
||||
parseMongoFindCommand,
|
||||
parseMongoGetIndexesCommand,
|
||||
parseMongoWriteCommand,
|
||||
parseMongoUseCommand,
|
||||
parseMongoVersionCommand,
|
||||
type MongoAggregateSafetyOptions,
|
||||
} from "@/lib/mongoShellCommand";
|
||||
import { redisCommandResultToQueryResult } from "@/lib/redisQueryResult";
|
||||
@@ -1589,7 +1591,7 @@ export const useQueryStore = defineStore("query", () => {
|
||||
if (mongoFind) {
|
||||
await connStore.ensureConnected(tab.connectionId);
|
||||
console.info("[DBX][executeTabSql:mongo-find:start]", { traceId, collection: mongoFind.collection });
|
||||
const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoFind.collection, mongoFind.skip, mongoFind.limit, mongoFind.filter, mongoFind.sort, executionId);
|
||||
const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoFind.collection, mongoFind.skip, mongoFind.limit, mongoFind.filter, mongoFind.projection, mongoFind.sort, executionId);
|
||||
console.info("[DBX][executeTabSql:mongo-find:done]", {
|
||||
traceId,
|
||||
rowCount: result.documents.length,
|
||||
@@ -1613,11 +1615,38 @@ export const useQueryStore = defineStore("query", () => {
|
||||
}
|
||||
return;
|
||||
}
|
||||
const mongoVersion = conn?.db_type === "mongodb" ? parseMongoVersionCommand(sql) : null;
|
||||
if (mongoVersion) {
|
||||
await connStore.ensureConnected(tab.connectionId);
|
||||
console.info("[DBX][executeTabSql:mongo-version:start]", { traceId });
|
||||
const version = await api.mongoServerVersion(tab.connectionId, tab.database, executionId);
|
||||
console.info("[DBX][executeTabSql:mongo-version:done]", {
|
||||
traceId,
|
||||
version,
|
||||
elapsed: elapsed(),
|
||||
});
|
||||
const current = tabs.value.find((t) => t.id === id);
|
||||
if (current?.executionId === executionId) {
|
||||
current.results = undefined;
|
||||
current.activeResultIndex = undefined;
|
||||
current.result = markQueryResultRowsRaw(mongoVersionToQueryResult(version, performance.now() - startedAt));
|
||||
touchResult(current);
|
||||
current.queryAnalysis = undefined;
|
||||
current.querySourceColumns = undefined;
|
||||
current.queryEditabilityReason = undefined;
|
||||
current.mongoEditTarget = undefined;
|
||||
current.tableMeta = undefined;
|
||||
current.resultBaseSql = options?.resultBaseSql ?? sql;
|
||||
current.resultSortedSql = options?.resultSortedSql;
|
||||
syncDisplayedResultRun(current, options?.resultBaseSql ?? sql);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const mongoCount = conn?.db_type === "mongodb" ? parseMongoCountDocumentsCommand(sql) : null;
|
||||
if (mongoCount) {
|
||||
await connStore.ensureConnected(tab.connectionId);
|
||||
console.info("[DBX][executeTabSql:mongo-count:start]", { traceId, collection: mongoCount.collection });
|
||||
const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoCount.collection, 0, 1, mongoCount.filter, undefined, executionId);
|
||||
const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoCount.collection, 0, 1, mongoCount.filter, undefined, undefined, executionId);
|
||||
console.info("[DBX][executeTabSql:mongo-count:done]", {
|
||||
traceId,
|
||||
total: result.total,
|
||||
|
||||
@@ -68,6 +68,7 @@
|
||||
"list_databases",
|
||||
"list_collections",
|
||||
"find_documents",
|
||||
"server_version",
|
||||
"insert_document",
|
||||
"update_document",
|
||||
"delete_document"
|
||||
|
||||
@@ -243,16 +243,18 @@ pub enum MongoAgentMethod {
|
||||
ListDatabases,
|
||||
ListCollections,
|
||||
FindDocuments,
|
||||
ServerVersion,
|
||||
InsertDocument,
|
||||
UpdateDocument,
|
||||
DeleteDocument,
|
||||
}
|
||||
|
||||
impl MongoAgentMethod {
|
||||
pub const ALL: [Self; 6] = [
|
||||
pub const ALL: [Self; 7] = [
|
||||
Self::ListDatabases,
|
||||
Self::ListCollections,
|
||||
Self::FindDocuments,
|
||||
Self::ServerVersion,
|
||||
Self::InsertDocument,
|
||||
Self::UpdateDocument,
|
||||
Self::DeleteDocument,
|
||||
@@ -263,6 +265,7 @@ impl MongoAgentMethod {
|
||||
Self::ListDatabases => "list_databases",
|
||||
Self::ListCollections => "list_collections",
|
||||
Self::FindDocuments => "find_documents",
|
||||
Self::ServerVersion => "server_version",
|
||||
Self::InsertDocument => "insert_document",
|
||||
Self::UpdateDocument => "update_document",
|
||||
Self::DeleteDocument => "delete_document",
|
||||
@@ -921,6 +924,13 @@ impl AgentDriverClient {
|
||||
self.call_mongo_method(MongoAgentMethod::FindDocuments, params).await
|
||||
}
|
||||
|
||||
pub async fn mongo_server_version<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
database: &str,
|
||||
) -> Result<T, String> {
|
||||
self.call_mongo_method(MongoAgentMethod::ServerVersion, mongo_database_params(database)).await
|
||||
}
|
||||
|
||||
pub async fn mongo_insert_document<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
params: Value,
|
||||
@@ -1461,6 +1471,7 @@ mod tests {
|
||||
assert_eq!(MongoAgentMethod::ListDatabases.as_str(), "list_databases");
|
||||
assert_eq!(MongoAgentMethod::ListCollections.as_str(), "list_collections");
|
||||
assert_eq!(MongoAgentMethod::FindDocuments.as_str(), "find_documents");
|
||||
assert_eq!(MongoAgentMethod::ServerVersion.as_str(), "server_version");
|
||||
assert_eq!(MongoAgentMethod::InsertDocument.as_str(), "insert_document");
|
||||
assert_eq!(MongoAgentMethod::UpdateDocument.as_str(), "update_document");
|
||||
assert_eq!(MongoAgentMethod::DeleteDocument.as_str(), "delete_document");
|
||||
@@ -1500,6 +1511,7 @@ mod tests {
|
||||
let _mongo_list_databases = AgentDriverClient::mongo_list_databases::<serde_json::Value>;
|
||||
let _mongo_list_collections = AgentDriverClient::mongo_list_collections::<serde_json::Value>;
|
||||
let _mongo_find_documents = AgentDriverClient::mongo_find_documents::<serde_json::Value>;
|
||||
let _mongo_server_version = AgentDriverClient::mongo_server_version::<serde_json::Value>;
|
||||
let _mongo_insert_document = AgentDriverClient::mongo_insert_document::<serde_json::Value>;
|
||||
let _mongo_update_document = AgentDriverClient::mongo_update_document::<serde_json::Value>;
|
||||
let _mongo_delete_document = AgentDriverClient::mongo_delete_document::<serde_json::Value>;
|
||||
|
||||
@@ -113,6 +113,17 @@ pub async fn test_connection(client: &Client, timeout: Duration, database: Optio
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn server_version(client: &Client, database: &str) -> Result<String, String> {
|
||||
let database = database.trim();
|
||||
let database = if database.is_empty() { "admin" } else { database };
|
||||
let result = client.database(database).run_command(doc! { "buildInfo": 1 }).await.map_err(|e| e.to_string())?;
|
||||
server_version_from_build_info(&result)
|
||||
}
|
||||
|
||||
fn server_version_from_build_info(result: &Document) -> Result<String, String> {
|
||||
result.get_str("version").map(str::to_string).map_err(|e| format!("MongoDB server version not found: {e}"))
|
||||
}
|
||||
|
||||
pub async fn list_databases(client: &Client) -> Result<Vec<String>, String> {
|
||||
client.list_database_names().await.map_err(|e| e.to_string())
|
||||
}
|
||||
@@ -193,6 +204,7 @@ pub async fn find_documents(
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
sort: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
@@ -212,6 +224,14 @@ pub async fn find_documents(
|
||||
};
|
||||
|
||||
let mut find = col.find(filter_doc).skip(skip).limit(limit);
|
||||
if let Some(p) = projection {
|
||||
if !p.trim().is_empty() {
|
||||
let json: serde_json::Value =
|
||||
serde_json::from_str(p).map_err(|e| format!("Invalid projection JSON: {e}"))?;
|
||||
let projection_doc = json_object_to_document(&json).map_err(|e| format!("Invalid projection: {e}"))?;
|
||||
find = find.projection(projection_doc);
|
||||
}
|
||||
}
|
||||
if let Some(s) = sort {
|
||||
if !s.trim().is_empty() {
|
||||
let json: serde_json::Value = serde_json::from_str(s).map_err(|e| format!("Invalid sort JSON: {e}"))?;
|
||||
@@ -844,6 +864,32 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_object_to_document_parses_find_projection() {
|
||||
let value = serde_json::json!({
|
||||
"title": 1,
|
||||
"_id": 0,
|
||||
});
|
||||
let doc = json_object_to_document(&value).unwrap();
|
||||
|
||||
assert!(matches!(doc.get("title"), Some(Bson::Int64(1))));
|
||||
assert!(matches!(doc.get("_id"), Some(Bson::Int64(0))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn server_version_from_build_info_reads_version_field() {
|
||||
let version = server_version_from_build_info(&doc! { "version": "4.4.29" }).unwrap();
|
||||
|
||||
assert_eq!(version, "4.4.29");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn server_version_from_build_info_rejects_missing_version() {
|
||||
let error = server_version_from_build_info(&doc! { "ok": 1 }).unwrap_err();
|
||||
|
||||
assert!(error.contains("MongoDB server version not found"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_object_to_document_parses_mongo_shell_isodate_strings() {
|
||||
let value = serde_json::json!({
|
||||
|
||||
@@ -123,6 +123,23 @@ pub async fn mongo_drop_collection_core(
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_server_version_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
) -> Result<String, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => mongo_driver::server_version(client, database).await,
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
client.mongo_server_version(database).await
|
||||
}
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn document_find_documents_core(
|
||||
state: &AppState,
|
||||
@@ -132,13 +149,14 @@ pub async fn document_find_documents_core(
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
sort: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::find_documents(client, database, collection, skip, limit, filter, sort).await
|
||||
mongo_driver::find_documents(client, database, collection, skip, limit, filter, projection, sort).await
|
||||
}
|
||||
PoolKind::Elasticsearch(client) => {
|
||||
let client = client.clone();
|
||||
@@ -153,16 +171,18 @@ pub async fn document_find_documents_core(
|
||||
}
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
client
|
||||
.mongo_find_documents(serde_json::json!({
|
||||
"database": database,
|
||||
"collection": collection,
|
||||
"skip": skip,
|
||||
"limit": limit,
|
||||
"filter": filter,
|
||||
"sort": sort,
|
||||
}))
|
||||
.await
|
||||
let mut params = serde_json::json!({
|
||||
"database": database,
|
||||
"collection": collection,
|
||||
"skip": skip,
|
||||
"limit": limit,
|
||||
"filter": filter,
|
||||
"sort": sort,
|
||||
});
|
||||
if let Some(projection) = projection {
|
||||
params["projection"] = serde_json::json!(projection);
|
||||
}
|
||||
client.mongo_find_documents(params).await
|
||||
}
|
||||
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
|
||||
}
|
||||
@@ -177,9 +197,11 @@ pub async fn mongo_find_documents_core(
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
sort: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
document_find_documents_core(state, connection_id, database, collection, skip, limit, filter, sort).await
|
||||
document_find_documents_core(state, connection_id, database, collection, skip, limit, filter, projection, sort)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn mongo_aggregate_documents_core(
|
||||
|
||||
@@ -1967,6 +1967,7 @@ async fn find_mongo_documents_for_transfer(
|
||||
offset,
|
||||
batch_size as i64,
|
||||
None,
|
||||
None,
|
||||
Some(r#"{"_id":1}"#),
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -365,6 +365,7 @@ async fn main() {
|
||||
.route("/mongo/drop-collection", post(routes::mongo::drop_collection))
|
||||
.route("/document-store/find-documents", post(routes::mongo::document_find_documents))
|
||||
.route("/mongo/find-documents", post(routes::mongo::find_documents))
|
||||
.route("/mongo/server-version", post(routes::mongo::server_version))
|
||||
.route("/mongo/aggregate-documents", post(routes::mongo::aggregate_documents))
|
||||
.route("/mongo/insert-document", post(routes::mongo::insert_document))
|
||||
.route("/mongo/insert-documents", post(routes::mongo::insert_documents))
|
||||
|
||||
@@ -73,10 +73,19 @@ pub struct MongoFindRequest {
|
||||
pub skip: Option<u64>,
|
||||
pub limit: Option<i64>,
|
||||
pub filter: Option<String>,
|
||||
pub projection: Option<String>,
|
||||
pub sort: Option<String>,
|
||||
pub execution_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoServerVersionRequest {
|
||||
pub connection_id: String,
|
||||
pub database: String,
|
||||
pub execution_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoAggregateRequest {
|
||||
@@ -213,6 +222,7 @@ pub async fn find_documents(
|
||||
req.skip.unwrap_or(0),
|
||||
req.limit.unwrap_or(50),
|
||||
req.filter.as_deref(),
|
||||
req.projection.as_deref(),
|
||||
req.sort.as_deref(),
|
||||
),
|
||||
)
|
||||
@@ -235,6 +245,7 @@ pub async fn document_find_documents(
|
||||
req.skip.unwrap_or(0),
|
||||
req.limit.unwrap_or(50),
|
||||
req.filter.as_deref(),
|
||||
req.projection.as_deref(),
|
||||
req.sort.as_deref(),
|
||||
),
|
||||
)
|
||||
@@ -242,6 +253,19 @@ pub async fn document_find_documents(
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
|
||||
}
|
||||
|
||||
pub async fn server_version(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<MongoServerVersionRequest>,
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
let result = run_cancellable(
|
||||
&state,
|
||||
req.execution_id.clone(),
|
||||
dbx_core::mongo_ops::mongo_server_version_core(&state.app, &req.connection_id, &req.database),
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
|
||||
}
|
||||
|
||||
pub async fn aggregate_documents(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<MongoAggregateRequest>,
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
parseMongoCountDocumentsCommand,
|
||||
parseMongoFindCommand,
|
||||
parseMongoGetIndexesCommand,
|
||||
parseMongoVersionCommand,
|
||||
parseMongoWriteCommand,
|
||||
} from "../../apps/desktop/src/lib/mongoShellCommand.ts";
|
||||
import { buildMongoUpdateDocument as buildMongoDocumentUpdate, formatMongoShellLiteral as formatMongoDocumentShellLiteral } from "../../apps/desktop/src/lib/mongoDocumentValues.ts";
|
||||
@@ -67,8 +68,26 @@ test("parseMongoFindCommand accepts single-quoted string values and unquoted sor
|
||||
assert.deepEqual(JSON.parse(command.sort || "{}"), { price: -1 });
|
||||
});
|
||||
|
||||
test("parseMongoFindCommand parses projection arguments", () => {
|
||||
const command = parseMongoFindCommand(`db.jobs.find({ status: "open" }, {
|
||||
title: 1,
|
||||
_id: 0
|
||||
}).sort({ title: 1 })`);
|
||||
assert.ok(command);
|
||||
assert.equal(command.collection, "jobs");
|
||||
assert.deepEqual(JSON.parse(command.filter), { status: "open" });
|
||||
assert.deepEqual(JSON.parse(command.projection || "{}"), { title: 1, _id: 0 });
|
||||
assert.deepEqual(JSON.parse(command.sort || "{}"), { title: 1 });
|
||||
});
|
||||
|
||||
test("parseMongoFindCommand rejects unsupported mongo shell commands", () => {
|
||||
assert.equal(parseMongoFindCommand("db.users.drop()"), null);
|
||||
assert.equal(parseMongoFindCommand("db.users.find({}, {}, { hint: { name: 1 } })"), null);
|
||||
});
|
||||
|
||||
test("parseMongoVersionCommand parses db.version", () => {
|
||||
assert.deepEqual(parseMongoVersionCommand("db.version();"), { kind: "version" });
|
||||
assert.equal(parseMongoVersionCommand("db.jobs.version()"), null);
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand accepts unquoted insert and update commands", () => {
|
||||
|
||||
@@ -626,9 +626,14 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
if (config.db_type === "mongodb") {
|
||||
const find = parseMongoFindCommand(sql);
|
||||
if (find) {
|
||||
const result = await withTimeout(mongoFindDocuments(config, find.collection, find.skip, find.limit, find.filter, find.sort), resolveTimeoutMs(options));
|
||||
const result = await withTimeout(mongoFindDocuments(config, find.collection, find.skip, find.limit, find.filter, find.projection, find.sort), resolveTimeoutMs(options));
|
||||
return mongoDocumentsToQueryResult(result.documents.slice(0, resolveMaxRows(options)), result.total);
|
||||
}
|
||||
const version = parseMongoVersionCommand(sql);
|
||||
if (version) {
|
||||
const result = await withTimeout(mongoServerVersion(config), resolveTimeoutMs(options));
|
||||
return { columns: ["version"], rows: [{ version: result }], row_count: 1 };
|
||||
}
|
||||
const count = parseMongoCountDocumentsCommand(sql);
|
||||
if (count) {
|
||||
const result = await withTimeout(mongoFindDocuments(config, count.collection, 0, 1, count.filter), resolveTimeoutMs(options));
|
||||
@@ -653,7 +658,7 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
const affected = await withTimeout(executeMongoWrite(config, write), resolveTimeoutMs(options));
|
||||
return { columns: [], rows: [], row_count: affected };
|
||||
}
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
}
|
||||
if (isDirectQueryType(config.db_type)) {
|
||||
return query(config, sql, undefined, options);
|
||||
@@ -848,7 +853,7 @@ export async function describeTable(config: ConnectionConfig, table: string, sch
|
||||
}));
|
||||
}
|
||||
|
||||
async function mongoFindDocuments(config: ConnectionConfig, collection: string, skip: number, limit: number, filter: string, sort?: string): Promise<MongoDocumentResult> {
|
||||
async function mongoFindDocuments(config: ConnectionConfig, collection: string, skip: number, limit: number, filter: string, projection?: string, sort?: string): Promise<MongoDocumentResult> {
|
||||
return bridgeDataRequest<MongoDocumentResult>("/data/mongo/find-documents", {
|
||||
connection_name: config.name,
|
||||
database: config.database || "",
|
||||
@@ -856,10 +861,18 @@ async function mongoFindDocuments(config: ConnectionConfig, collection: string,
|
||||
skip,
|
||||
limit,
|
||||
filter,
|
||||
projection,
|
||||
sort,
|
||||
});
|
||||
}
|
||||
|
||||
async function mongoServerVersion(config: ConnectionConfig): Promise<string> {
|
||||
return bridgeDataRequest<string>("/data/mongo/server-version", {
|
||||
connection_name: config.name,
|
||||
database: config.database || "",
|
||||
});
|
||||
}
|
||||
|
||||
async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise<number> {
|
||||
if (command.kind === "insert") {
|
||||
const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/insert-documents", {
|
||||
@@ -901,7 +914,7 @@ async function mongoAggregateDocuments(config: ConnectionConfig, collection: str
|
||||
});
|
||||
}
|
||||
|
||||
export function mongoDocumentsToQueryResult(documents: unknown[], total: number): QueryResult {
|
||||
export function mongoDocumentsToQueryResult(documents: unknown[], _total: number): QueryResult {
|
||||
const columns: string[] = [];
|
||||
for (const doc of documents) {
|
||||
if (isRecord(doc)) {
|
||||
@@ -951,6 +964,7 @@ export function inferMongoColumns(documents: unknown[]): ColumnInfo[] {
|
||||
interface MongoFindCommand {
|
||||
collection: string;
|
||||
filter: string;
|
||||
projection?: string;
|
||||
skip: number;
|
||||
limit: number;
|
||||
sort?: string;
|
||||
@@ -980,8 +994,15 @@ export function parseMongoFindCommand(input: string): MongoFindCommand | null {
|
||||
const findCloseIndex = findMatchingParen(source, findOpenIndex);
|
||||
if (findCloseIndex < 0) return null;
|
||||
const findArgs = splitTopLevel(source.slice(findOpenIndex + 1, findCloseIndex));
|
||||
if (findArgs.length > 2 && findArgs.slice(2).some((arg) => arg.trim())) return null;
|
||||
const filter = normalizeJsonArgument(findArgs[0] || "{}");
|
||||
if (!filter) return null;
|
||||
let projection: string | undefined;
|
||||
if (findArgs[1]?.trim()) {
|
||||
const parsedProjection = normalizeJsonArgument(findArgs[1]);
|
||||
if (!parsedProjection) return null;
|
||||
projection = parsedProjection;
|
||||
}
|
||||
const chain = source.slice(findCloseIndex + 1).trim();
|
||||
if (chain && !chain.startsWith(".")) return null;
|
||||
const sortArg = readChainedCallArgument(chain, "sort");
|
||||
@@ -994,7 +1015,12 @@ export function parseMongoFindCommand(input: string): MongoFindCommand | null {
|
||||
const skip = readChainedIntegerArgument(chain, "skip", 0);
|
||||
const limit = readChainedIntegerArgument(chain, "limit", MAX_ROWS);
|
||||
if (skip === null || limit === null) return null;
|
||||
return { collection: target.collection, filter, skip, limit, sort };
|
||||
return { collection: target.collection, filter, ...(projection ? { projection } : {}), skip, limit, sort };
|
||||
}
|
||||
|
||||
export function parseMongoVersionCommand(input: string): boolean {
|
||||
const source = input.trim().replace(/;$/, "").trim();
|
||||
return /^db\s*\.\s*version\s*\(\s*\)$/i.test(source);
|
||||
}
|
||||
|
||||
export function parseMongoCountDocumentsCommand(input: string): MongoCountDocumentsCommand | null {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { ConnectionConfig } from "./connections.js";
|
||||
import type { TableInfo, ColumnInfo, QueryOptions, QueryResult } from "./database.js";
|
||||
import { collectionListToTableInfos, evaluateMongoAggregateSafety, evaluateMongoWriteSafety, inferMongoColumns, mongoDocumentsToQueryResult, parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, parseMongoGetIndexesCommand, parseMongoWriteCommand, type CollectionInfo, type MongoWriteCommand } from "./database.js";
|
||||
import { collectionListToTableInfos, evaluateMongoAggregateSafety, evaluateMongoWriteSafety, inferMongoColumns, mongoDocumentsToQueryResult, parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, parseMongoGetIndexesCommand, parseMongoVersionCommand, parseMongoWriteCommand, type CollectionInfo, type MongoWriteCommand } from "./database.js";
|
||||
import type { RedisCommandOptions, RedisCommandResult } from "./redis-command.js";
|
||||
import { sqlSafetyFromEnv } from "./sql-safety.js";
|
||||
|
||||
@@ -140,12 +140,24 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
skip: find.skip,
|
||||
limit: find.limit,
|
||||
filter: find.filter,
|
||||
projection: find.projection,
|
||||
sort: find.sort,
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { documents: unknown[]; total: number };
|
||||
return mongoDocumentsToQueryResult(result.documents.slice(0, options?.maxRows ?? result.documents.length), result.total);
|
||||
}
|
||||
if (parseMongoVersionCommand(sql)) {
|
||||
const res = await apiFetch("/api/mongo/server-version", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({
|
||||
connectionId: config.id,
|
||||
database: config.database || "",
|
||||
}),
|
||||
});
|
||||
const version = (await res.json()) as string;
|
||||
return { columns: ["version"], rows: [{ version }], row_count: 1 };
|
||||
}
|
||||
const count = parseMongoCountDocumentsCommand(sql);
|
||||
if (count) {
|
||||
const res = await apiFetch("/api/mongo/find-documents", {
|
||||
@@ -201,7 +213,7 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
|
||||
const affected = await executeMongoWrite(config, write);
|
||||
return { columns: [], rows: [], row_count: affected };
|
||||
}
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
throw new Error("Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.version(), db.projects.countDocuments({}), db.projects.getIndexes(), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})");
|
||||
}
|
||||
const res = await apiFetch("/api/query/execute", {
|
||||
method: "POST",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "vitest";
|
||||
import { executeQuery, inferMongoColumns, mongoAggregateWriteStage, mongoDocumentsToQueryResult, parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, parseMongoGetIndexesCommand, parseMongoWriteCommand } from "../src/database.js";
|
||||
import { executeQuery, inferMongoColumns, mongoAggregateWriteStage, mongoDocumentsToQueryResult, parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, parseMongoGetIndexesCommand, parseMongoVersionCommand, parseMongoWriteCommand } from "../src/database.js";
|
||||
|
||||
test("parseMongoFindCommand accepts shell-style find commands", () => {
|
||||
assert.deepEqual(parseMongoFindCommand('db.getCollection("operation_logs").find({"level":"info"}).sort({"ts":-1}).skip(5).limit(10)'), {
|
||||
@@ -36,6 +36,20 @@ test("parseMongoFindCommand accepts Compass-style unquoted keys and ObjectId", (
|
||||
assert.deepEqual(JSON.parse(command.filter), { _id: { $oid: "6a045a92d2971e44243771a1" } });
|
||||
});
|
||||
|
||||
test("parseMongoFindCommand accepts projection arguments", () => {
|
||||
const command = parseMongoFindCommand("db.jobs.find({status: 'open'}, {title: 1, _id: 0}).sort({title: 1})");
|
||||
assert.ok(command);
|
||||
assert.equal(command.collection, "jobs");
|
||||
assert.deepEqual(JSON.parse(command.filter), { status: "open" });
|
||||
assert.deepEqual(JSON.parse(command.projection || "{}"), { title: 1, _id: 0 });
|
||||
assert.deepEqual(JSON.parse(command.sort || "{}"), { title: 1 });
|
||||
});
|
||||
|
||||
test("parseMongoVersionCommand accepts db.version", () => {
|
||||
assert.equal(parseMongoVersionCommand("db.version();"), true);
|
||||
assert.equal(parseMongoVersionCommand("db.jobs.version()"), false);
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand accepts unquoted update operator keys", () => {
|
||||
assert.deepEqual(parseMongoWriteCommand("db.projects.updateOne({_id: ObjectId('507f1f77bcf86cd799439011')}, {$set: {name: 'next'}})"), {
|
||||
kind: "update",
|
||||
|
||||
@@ -51,9 +51,16 @@ struct MongoFindDocumentsRequest {
|
||||
skip: Option<u64>,
|
||||
limit: Option<i64>,
|
||||
filter: Option<String>,
|
||||
projection: Option<String>,
|
||||
sort: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct MongoServerVersionRequest {
|
||||
connection_name: String,
|
||||
database: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct MongoAggregateDocumentsRequest {
|
||||
connection_name: String,
|
||||
@@ -157,6 +164,8 @@ pub fn start(app_handle: AppHandle, state: Arc<AppState>) {
|
||||
handle_mongo_list_collections_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/find-documents") {
|
||||
handle_mongo_find_documents_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/server-version") {
|
||||
handle_mongo_server_version_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/aggregate-documents") {
|
||||
handle_mongo_aggregate_documents_data(&st, body, &mut stream).await;
|
||||
} else if first_line.starts_with("POST /data/mongo/insert-documents") {
|
||||
@@ -410,6 +419,7 @@ async fn handle_mongo_find_documents_data(state: &Arc<AppState>, body: &str, str
|
||||
req.skip.unwrap_or(0),
|
||||
req.limit.unwrap_or(100),
|
||||
req.filter.as_deref(),
|
||||
req.projection.as_deref(),
|
||||
req.sort.as_deref(),
|
||||
)
|
||||
.await
|
||||
@@ -419,6 +429,25 @@ async fn handle_mongo_find_documents_data(state: &Arc<AppState>, body: &str, str
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_mongo_server_version_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
|
||||
let req: MongoServerVersionRequest = match serde_json::from_str(body) {
|
||||
Ok(r) => r,
|
||||
Err(_) => {
|
||||
respond_error(stream, "400 Bad Request", "Invalid JSON").await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let Some((pool_key, database, _connection_id)) =
|
||||
resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
match dbx_core::mongo_ops::mongo_server_version_core(state, &pool_key, &database).await {
|
||||
Ok(version) => respond_json(stream, &version).await,
|
||||
Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_mongo_aggregate_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
|
||||
let req: MongoAggregateDocumentsRequest = match serde_json::from_str(body) {
|
||||
Ok(r) => r,
|
||||
|
||||
@@ -81,6 +81,7 @@ pub async fn mongo_find_documents(
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
filter: Option<String>,
|
||||
projection: Option<String>,
|
||||
sort: Option<String>,
|
||||
execution_id: Option<String>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
@@ -96,6 +97,7 @@ pub async fn mongo_find_documents(
|
||||
skip,
|
||||
limit,
|
||||
filter.as_deref(),
|
||||
projection.as_deref(),
|
||||
sort.as_deref(),
|
||||
),
|
||||
)
|
||||
@@ -112,6 +114,7 @@ pub async fn document_find_documents(
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
filter: Option<String>,
|
||||
projection: Option<String>,
|
||||
sort: Option<String>,
|
||||
execution_id: Option<String>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
@@ -127,12 +130,25 @@ pub async fn document_find_documents(
|
||||
skip,
|
||||
limit,
|
||||
filter.as_deref(),
|
||||
projection.as_deref(),
|
||||
sort.as_deref(),
|
||||
),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn mongo_server_version(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
connection_id: String,
|
||||
database: String,
|
||||
execution_id: Option<String>,
|
||||
) -> Result<String, String> {
|
||||
let app = state.inner().clone();
|
||||
run_cancellable(&app, execution_id, dbx_core::mongo_ops::mongo_server_version_core(&app, &connection_id, &database))
|
||||
.await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn mongo_aggregate_documents(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
|
||||
@@ -846,6 +846,7 @@ pub fn run() {
|
||||
commands::mongo_cmd::mongo_drop_collection,
|
||||
commands::mongo_cmd::document_find_documents,
|
||||
commands::mongo_cmd::mongo_find_documents,
|
||||
commands::mongo_cmd::mongo_server_version,
|
||||
commands::mongo_cmd::mongo_aggregate_documents,
|
||||
commands::mongo_cmd::mongo_insert_document,
|
||||
commands::mongo_cmd::mongo_insert_documents,
|
||||
|
||||
Reference in New Issue
Block a user