drasi-source-sqlite 0.1.3

SQLite source plugin for Drasi
Documentation

SQLite Source

Overview

drasi-source-sqlite is a protocol/local source that owns an embedded SQLite connection and emits Drasi SourceChange events for row-level INSERT, UPDATE, and DELETE activity.

Key capabilities:

  • Real-time CDC using rusqlite preupdate_hook
  • Transaction-aware dispatch (commit_hook flush, rollback_hook discard)
  • Savepoint-aware buffering for partial rollback support
  • Optional table filtering and explicit table key configuration
  • Optional REST API for table CRUD and transactional batch operations
  • Works with file-backed and in-memory SQLite databases

Quick Start

use drasi_source_sqlite::{SqliteSource, TableKeyConfig, RestApiConfig};

let source = SqliteSource::builder("sqlite-source")
    .with_path("data/example.db")
    .with_table_keys(vec![TableKeyConfig {
        table: "sensors".to_string(),
        key_columns: vec!["id".to_string()],
    }])
    .with_rest_api(RestApiConfig {
        host: "127.0.0.1".to_string(),
        port: 9100,
    })
    .build()?;

SQL Handle API

Use source.handle() to execute statements through the source-owned SQLite connection:

let handle = source.handle();
handle.execute("CREATE TABLE sensors(id INTEGER PRIMARY KEY, name TEXT, temp REAL)").await?;
handle.execute("INSERT INTO sensors(id, name, temp) VALUES (1, 'sensor-a', 31.5)").await?;

handle.transaction(|tx| async move {
    tx.execute("INSERT INTO sensors(id, name, temp) VALUES (2, 'sensor-b', 30.0)").await?;
    tx.execute("UPDATE sensors SET temp = 32.1 WHERE id = 1").await?;
    Ok(())
}).await?;

REST API (Optional)

When with_rest_api() is configured:

  • GET /health
  • GET /api/tables
  • GET /api/tables/{table}
  • GET /api/tables/{table}/{id}
  • POST /api/tables/{table}
  • PUT /api/tables/{table}/{id}
  • DELETE /api/tables/{table}/{id}
  • POST /api/batch (single SQLite transaction for all operations)

Configuration Options

Property Type Default Description
path String? null SQLite file path. Omit for in-memory database
tables Vec<String>? null Optional table allow-list. null means all user tables
tableKeys Vec<TableKeyConfig> [] Manual primary key configuration (see below)
restApi RestApiConfig? null Optional REST API configuration

TableKeyConfig

Field Type Description
table String Table name
keyColumns Vec<String> Column names to use as primary key

Configured table keys override automatically detected primary keys from PRAGMA table_info. If no configured or detected keys exist, rowid is used as the fallback for streaming events.

RestApiConfig

Field Type Default Description
host String "0.0.0.0" Address to bind
port u16 8080 Port to bind

Testing

From this crate directory:

cargo test
cargo test -- --ignored --nocapture
cargo clippy --all-targets -- -D warnings

The ignored integration tests validate:

  • handle-driven INSERT/UPDATE/DELETE flow
  • REST CRUD and transactional batch behavior
  • bootstrap delivery into query results
  • multi-table query routing

Limitations

  • CDC only captures writes executed through the source-owned connection.
  • DDL (CREATE/ALTER/DROP TABLE) is not emitted as SourceChange.
  • BLOB values are emitted as base64-encoded strings.

Troubleshooting

  • source ... is not running: call core.start() before using SqliteSourceHandle.
  • no events after writes: verify writes go through the source handle or REST API, not an external SQLite connection.
  • REST 400 responses: table/column identifiers failed validation.