Add endpoints for upserting, retrieving, and

listing database rows

core/bin/dust_api.rs

@@ -1870,6 +1870,140 @@ async fn databases_tables_list(
     }
 }
 
+#[derive(serde::Deserialize)]
+struct DatabasesRowsUpsertPayload {
+    contents: HashMap<String, serde_json::Value>,
+    truncate: Option<bool>,
+}
+
+async fn databases_rows_upsert(
+    extract::Path((project_id, data_source_id, database_id, table_id)): extract::Path<(
+        i64,
+        String,
+        String,
+        String,
+    )>,
+    extract::Json(payload): extract::Json<DatabasesRowsUpsertPayload>,
+    extract::Extension(state): extract::Extension<Arc<APIState>>,
+) -> (StatusCode, Json<APIResponse>) {
+    let truncate = match payload.truncate {
+        Some(v) => v,
+        None => false,
+    };
+    let project = project::Project::new_from_id(project_id);
+
+    match state
+        .store
+        .batch_upsert_database_rows(
+            &project,
+            &data_source_id,
+            &database_id,
+            &table_id,
+            &payload.contents,
+            truncate,
+        )
+        .await
+    {
+        Err(e) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "internal_server_error",
+            "Failed to upsert database rows",
+            Some(e),
+        ),
+        Ok(()) => (
+            StatusCode::OK,
+            Json(APIResponse {
+                error: None,
+                response: Some(json!({
+                    "success": true
+                })),
+            }),
+        ),
+    }
+}
+
+async fn databases_rows_retrieve(
+    extract::Path((project_id, data_source_id, database_id, table_id, row_id)): extract::Path<(
+        i64,
+        String,
+        String,
+        String,
+        String,
+    )>,
+
+    extract::Extension(state): extract::Extension<Arc<APIState>>,
+) -> (StatusCode, Json<APIResponse>) {
+    let project = project::Project::new_from_id(project_id);
+
+    match state
+        .store
+        .load_database_row(&project, &data_source_id, &database_id, &table_id, &row_id)
+        .await
+    {
+        Err(e) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "internal_server_error",
+            "Failed to upsert database rows",
+            Some(e),
+        ),
+        Ok(row) => (
+            StatusCode::OK,
+            Json(APIResponse {
+                error: None,
+                response: Some(json!({
+                    "row": row
+                })),
+            }),
+        ),
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct DatabasesRowsListQuery {
+    offset: usize,
+    limit: usize,
+    table_id: Option<String>,
+}
+
+async fn databases_rows_list(
+    extract::Path((project_id, data_source_id, database_id)): extract::Path<(i64, String, String)>,
+    extract::Query(query): extract::Query<DatabasesRowsListQuery>,
+    extract::Extension(state): extract::Extension<Arc<APIState>>,
+) -> (StatusCode, Json<APIResponse>) {
+    let project = project::Project::new_from_id(project_id);
+
+    match state
+        .store
+        .list_database_rows(
+            &project,
+            &data_source_id,
+            &database_id,
+            &query.table_id,
+            Some((query.limit, query.offset)),
+        )
+        .await
+    {
+        Err(e) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "internal_server_error",
+            "Failed to list database rows",
+            Some(e),
+        ),
+        Ok((rows, total)) => (
+            StatusCode::OK,
+            Json(APIResponse {
+                error: None,
+                response: Some(json!({
+                    "rows": rows,
+                    "offset": query.offset,
+                    "limit": query.limit,
+                    "total": total,
+                })),
+            }),
+        ),
+    }
+}
+
 // Misc
 
 #[derive(serde::Deserialize)]
@@ -2075,6 +2209,18 @@ fn main() {
             "/projects/:project_id/data_sources/:data_source_id/databases/:database_id/tables",
             get(databases_tables_list),
         )
+        .route(
+            "/projects/:project_id/data_sources/:data_source_id/databases/:database_id/tables/:table_id",
+            post(databases_rows_upsert),
+        )
+        .route(
+            "/projects/:project_id/data_sources/:data_source_id/databases/:database_id/tables/:table_id/rows/:row_id",
+            get(databases_rows_retrieve),
+        )
+        .route(
+            "/projects/:project_id/data_sources/:data_source_id/databases/:database_id/rows",
+            get(databases_rows_list),
+        )
         // Misc
         .route("/tokenize", post(tokenize))
 

core/src/stores/postgres.rs

@@ -2259,7 +2259,7 @@ impl Store for PostgresStore {
         project: &Project,
         data_source_id: &str,
         database_id: &str,
-        table_id: Option<&str>,
+        table_id: &Option<String>,
         limit_offset: Option<(usize, usize)>,
     ) -> Result<(Vec<DatabaseRow>, usize)> {
         let project_id = project.project_id();

core/src/stores/store.rs

@@ -218,7 +218,7 @@ pub trait Store {
         project: &Project,
         data_source_id: &str,
         database_id: &str,
-        table_id: Option<&str>,
+        table_id: &Option<String>,
         limit_offset: Option<(usize, usize)>,
     ) -> Result<(Vec<DatabaseRow>, usize)>;
 

front/lib/core_api.ts

@@ -127,6 +127,13 @@ type CoreAPIDatabaseTable = {
   description: string;
 };
 
+type CoreAPIDatabaseRow = {
+  created: number;
+  table_id: string;
+  row_id: string;
+  content: Record<string, unknown>;
+};
+
 export const CoreAPI = {
   async createProject(): Promise<CoreAPIResponse<{ project: Project }>> {
     const response = await fetch(`${CORE_API}/projects`, {
@@ -953,6 +960,93 @@ export const CoreAPI = {
 
     return _resultFromResponse(response);
   },
+
+  async upsertDatabaseRows({
+    projectId,
+    dataSourceName,
+    databaseId,
+    tableId,
+    contents,
+    truncate,
+  }: {
+    projectId: string;
+    dataSourceName: string;
+    databaseId: string;
+    tableId: string;
+    contents: Record<string, CoreAPIDatabaseRow["content"]>;
+    truncate?: boolean;
+  }): Promise<CoreAPIResponse<{ success: true }>> {
+    const response = await fetch(
+      `${CORE_API}/projects/${projectId}/data_sources/${dataSourceName}/databases/${databaseId}/tables/${tableId}/rows`,
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          contents: contents,
+          truncate: truncate || false,
+        }),
+      }
+    );
+
+    return _resultFromResponse(response);
+  },
+
+  async getDatabaseRow({
+    projectId,
+    dataSourceName,
+    databaseId,
+    tableId,
+    rowId,
+  }: {
+    projectId: string;
+    dataSourceName: string;
+    databaseId: string;
+    tableId: string;
+    rowId: string;
+  }): Promise<CoreAPIResponse<{ row: CoreAPIDatabaseRow }>> {
+    const response = await fetch(
+      `${CORE_API}/projects/${projectId}/data_sources/${dataSourceName}/databases/${databaseId}/tables/${tableId}/rows/${rowId}`,
+      {
+        method: "GET",
+      }
+    );
+
+    return _resultFromResponse(response);
+  },
+
+  async getDatabaseRows({
+    projectId,
+    dataSourceName,
+    databaseId,
+    tableId,
+    limit,
+    offset,
+  }: {
+    projectId: string;
+    dataSourceName: string;
+    databaseId: string;
+    tableId: string;
+    limit: number;
+    offset: number;
+  }): Promise<
+    CoreAPIResponse<{
+      rows: CoreAPIDatabaseRow[];
+      offset: number;
+      limit: number;
+      total: number;
+    }>
+  > {
+    const response = await fetch(
+      `${CORE_API}/projects/${projectId}/data_sources/${dataSourceName}/databases/${databaseId}/tables/${tableId}/rows?limit=${limit}&offset=${offset}`,
+      {
+        method: "GET",
+      }
+    );
+
+    return _resultFromResponse(response);
+  },
 };
 
 async function _resultFromResponse<T>(