Feat: monetdb datasource

ai/requirements.txt

@@ -18,4 +18,5 @@ redshift-connector==2.0.917
 mysqlclient==2.2.4
 trino==0.329.0
 snowflake-connector-python==3.12.2
-snowflake-sqlalchemy==1.6.1
+snowflake-sqlalchemy==1.6.1
+sqlalchemy_monetdb==2.0.0

apps/api/jupyter-requirements.txt

@@ -56,4 +56,5 @@ openpyxl==3.1.2
 mysqlclient==2.2.4
 pymongo==4.8.0
 snowflake-connector-python==3.12.2
-snowflake-sqlalchemy==1.6.1
+snowflake-sqlalchemy==1.6.1
+sqlalchemy_monetdb==2.0.0

apps/api/src/auth/token.ts

@@ -318,5 +318,9 @@ export const isAuthorizedForDataSource = async (
       const result = await prisma().snowflakeDataSource.findFirst(query)
       return result !== null
     }
+    case 'monetdb': {
+      const result = await prisma().monetDBDataSource.findFirst(query)
+      return result !== null
+    }
   }
 }

apps/api/src/datasources/index.ts

@@ -8,6 +8,7 @@ import * as mysql from './mysql.js'
 import * as trino from './trino.js'
 import * as sqlserver from './sqlserver.js'
 import * as snowflake from './snowflake.js'
+import * as monetdb from "./monetdb.js"
 import { DataSourceConnectionError } from '@briefer/types'
 
 export async function ping(ds: DataSource): Promise<DataSource> {
@@ -40,6 +41,9 @@ export async function ping(ds: DataSource): Promise<DataSource> {
     case 'snowflake':
       result = await snowflake.ping(ds.data)
       break
+    case "monetdb":
+      result = await monetdb.ping(ds.data)
+      break
   }
 
   return result
@@ -78,5 +82,7 @@ export async function updateConnStatus(
       return trino.updateConnStatus(ds.data, status)
     case 'snowflake':
       return snowflake.updateConnStatus(ds.data, status)
+    case "monetdb":
+      return monetdb.updateConnStatus(ds.data, status)
   }
 }

apps/api/src/datasources/monetdb.ts

@@ -0,0 +1,46 @@
+import { config } from '../config/index.js'
+import prisma, { DataSource, MonetDBDataSource } from '@briefer/database'
+import { DataSourceStatus } from './index.js'
+import { pingMonetDb } from '../python/query/monetdb.js'
+
+export async function ping(ds: MonetDBDataSource): Promise<DataSource> {
+  const lastConnection = new Date()
+  const err = await pingMonetDb(ds, config().DATASOURCES_ENCRYPTION_KEY)
+
+  if (!err) {
+    return updateConnStatus(ds, {
+      connStatus: 'online',
+      lastConnection,
+    })
+  }
+
+  return updateConnStatus(ds, { connStatus: 'offline', connError: err })
+}
+
+export async function updateConnStatus(
+  ds: MonetDBDataSource,
+  status: DataSourceStatus
+): Promise<DataSource> {
+  const newDs = await prisma().monetDBDataSource.update({
+    where: { id: ds.id },
+    data: {
+      connStatus: status.connStatus,
+      lastConnection:
+        status.connStatus === 'online' ? status.lastConnection : undefined,
+      connError:
+        status.connStatus === 'offline'
+          ? JSON.stringify(status.connError)
+          : undefined,
+    },
+  })
+
+  return {
+    type: "monetdb",
+    data: {
+      ...ds,
+      connStatus: newDs.connStatus,
+      lastConnection: newDs.lastConnection?.toISOString() ?? null,
+      connError: newDs.connError,
+    },
+  }
+}

apps/api/src/datasources/structure.ts

@@ -23,6 +23,7 @@ import { getOracleSchema } from '../python/query/oracle.js'
 import { getBigQuerySchema } from '../python/query/bigquery.js'
 import { getTrinoSchema } from '../python/query/trino.js'
 import { getSnowflakeSchema } from '../python/query/snowflake.js'
+import { getMonetDBSchema } from '../python/query/monetdb.js'
 import { getAthenaSchema } from './athena.js'
 import { getMySQLSchema } from './mysql.js'
 import { z } from 'zod'
@@ -127,6 +128,14 @@ async function getFromCache(
         })
       ).structure
       break
+    case "monetdb":
+      encrypted = (
+        await prisma().monetDBDataSource.findUniqueOrThrow({
+          where: { id: dataSourceId },
+          select: { structure: true },
+        })
+      ).structure
+      break
   }
 
   if (encrypted === null) {
@@ -307,6 +316,13 @@ async function _refreshDataSourceStructure(
           onTable
         )
         break
+      case "monetdb":
+        finalStructure = await getMonetDBSchema(
+          dataSource.config.data,
+          config().DATASOURCES_ENCRYPTION_KEY,
+          onTable
+        )
+        break
     }
 
     await updateQueue.onIdle()
@@ -461,6 +477,12 @@ async function persist(ds: APIDataSource): Promise<APIDataSource> {
           data: { structure },
         })
         return ds
+      case "monetdb":
+        await prisma().monetDBDataSource.update({
+          where: { id: ds.config.data.id },
+          data: { structure },
+        })
+        return ds
     }
   })()
 }

apps/api/src/python/query/index.ts

@@ -21,6 +21,7 @@ import { makeSnowflakeQuery } from './snowflake.js'
 import { updateConnStatus } from '../../datasources/index.js'
 import { getJupyterManager } from '../../jupyter/index.js'
 import { makeSQLServerQuery } from './sqlserver.js'
+import { makeMonetDBQuery } from './monetdb.js'
 
 export async function makeSQLQuery(
   workspaceId: string,
@@ -144,6 +145,18 @@ export async function makeSQLQuery(
         onProgress
       )
       break
+    case "monetdb":
+      result = await makeMonetDBQuery(
+        workspaceId,
+        sessionId,
+        queryId,
+        dataframeName,
+        datasource.data,
+        encryptionKey,
+        sql,
+        onProgress
+      )
+      break
   }
 
   result[0].then(async (r) => {

apps/api/src/python/query/monetdb.ts

@@ -0,0 +1,62 @@
+import { v4 as uuidv4 } from 'uuid'
+import {
+  MonetDBDataSource,
+  getDatabaseURL,
+} from '@briefer/database'
+import {
+  DataSourceStructure,
+  RunQueryResult,
+  SuccessRunQueryResult,
+} from '@briefer/types'
+import {
+  getSQLAlchemySchema,
+  makeSQLAlchemyQuery,
+  pingSQLAlchemy,
+} from './sqlalchemy.js'
+import { OnTable } from '../../datasources/structure.js'
+
+export async function makeMonetDBQuery(
+  workspaceId: string,
+  sessionId: string,
+  queryId: string,
+  dataframeName: string,
+  datasource: MonetDBDataSource,
+  encryptionKey: string,
+  sql: string,
+  onProgress: (result: SuccessRunQueryResult) => void
+): Promise<[Promise<RunQueryResult>, () => Promise<void>]> {
+  const databaseUrl = await getDatabaseURL(
+    { type: "monetdb", data: datasource },
+    encryptionKey
+  )
+
+  const jobId = uuidv4()
+  const query = `${sql}  -- Briefer jobId: ${jobId}`
+
+  return makeSQLAlchemyQuery(
+    workspaceId,
+    sessionId,
+    dataframeName,
+    databaseUrl,
+    "monetdb",
+    jobId,
+    query,
+    queryId,
+    onProgress
+  )
+}
+
+export function pingMonetDb(
+  ds: MonetDBDataSource,
+  encryptionKey: string
+): Promise<null | Error> {
+  return pingSQLAlchemy({ type: "monetdb", data: ds }, encryptionKey, null)
+}
+
+export function getMonetDBSchema(
+  ds: MonetDBDataSource,
+  encryptionKey: string,
+  onTable: OnTable
+): Promise<DataSourceStructure> {
+  return getSQLAlchemySchema({ type: "monetdb", data: ds }, encryptionKey, null, onTable)
+}

apps/api/src/python/query/sqlalchemy.ts

@@ -24,7 +24,8 @@ export async function makeSQLAlchemyQuery(
     | 'psql'
     | 'redshift'
     | 'trino'
-    | 'snowflake',
+    | 'snowflake'
+    | "monetdb",
   jobId: string,
   query: string,
   queryId: string,

apps/api/src/python/writeback/index.ts

@@ -54,6 +54,7 @@ export async function writeback(
     case 'oracle':
     case 'athena':
     case 'snowflake':
+    case "monetdb":
     case 'trino':
       throw new Error(`${datasource.type} writeback not implemented`)
   }

apps/api/src/v1/workspaces/workspace/data-sources/data-source.ts

@@ -29,6 +29,9 @@ import {
   getSnowflakeDataSource,
   updateSnowflakeDataSource,
   deleteSnowflakeDataSource,
+  getMonetDbDataSource,
+  updateMonedDbDataSource,
+  deleteMonetDbDataSource
 } from '@briefer/database'
 import { z } from 'zod'
 import { getParam } from '../../../../utils/express.js'
@@ -42,6 +45,7 @@ import * as mysql from '../../../../datasources/mysql.js'
 import * as sqlserver from '../../../../datasources/sqlserver.js'
 import * as trino from '../../../../datasources/trino.js'
 import * as snowflake from '../../../../datasources/snowflake.js'
+import * as monetDb from "../../../../datasources/monetdb.js"
 import { ping } from '../../../../datasources/index.js'
 import { fetchDataSourceStructure } from '../../../../datasources/structure.js'
 import {
@@ -148,6 +152,21 @@ const dataSourceRouter = (socketServer: IOServer) => {
         notes: z.string(),
       }),
     }),
+    z.object({
+      type: z.literal('monetdb'),
+      data: z.object({
+        id: z.string().min(1),
+        name: z.string().min(1),
+        host: z.string().min(1),
+        port: z.string().min(1),
+        database: z.string().min(1),
+        username: z.string().min(1),
+        password: z.string(),
+        notes: z.string(),
+        readOnly: z.boolean(),
+        cert: z.string().optional(),
+      }),
+    }),
   ])
 
   router.put('/', async (req, res) => {
@@ -171,6 +190,7 @@ const dataSourceRouter = (socketServer: IOServer) => {
         getSQLServerDataSource(workspaceId, dataSourceId),
         getTrinoDataSource(workspaceId, dataSourceId),
         getSnowflakeDataSource(workspaceId, dataSourceId),
+        getMonetDbDataSource(workspaceId, dataSourceId),
       ])
     ).find((e) => e !== null)
     if (!existingDb) {
@@ -325,6 +345,20 @@ const dataSourceRouter = (socketServer: IOServer) => {
         await snowflake.ping(snowflakeDs)
         break
       }
+      case "monetdb": {
+        const monetDbDs = await updateMonedDbDataSource(
+          {
+            ...data.data,
+            id: dataSourceId,
+            password:
+              data.data.password === '' ? undefined : data.data.password,
+          },
+          config().DATASOURCES_ENCRYPTION_KEY
+        )
+
+        await monetDb.ping(monetDbDs)
+        break
+      }
     }
 
     const ds = await getDatasource(workspaceId, dataSourceId, data.type)
@@ -473,6 +507,19 @@ const dataSourceRouter = (socketServer: IOServer) => {
         }
       }
 
+      const targetMonetDb = await recoverFromNotFound(
+        deleteMonetDbDataSource(workspaceId, targetId)
+      )
+      if(targetMonetDb) {
+        return {
+          status: 200,
+          payload: {
+            type: 'monetdb',
+            data: targetMonetDb,
+          },
+        }
+      }
+
       return { status: 404 }
     }
 
@@ -497,6 +544,7 @@ const dataSourceRouter = (socketServer: IOServer) => {
       z.literal('trino'),
       z.literal('sqlserver'),
       z.literal('snowflake'),
+      z.literal("monetdb"),
     ]),
   })
   router.post('/ping', async (req, res) => {
@@ -520,6 +568,7 @@ const dataSourceRouter = (socketServer: IOServer) => {
         return
       }
 
+
       const dsConfig = await ping(dataSource)
 
       await broadcastDataSource(socketServer, {