PULSE DB

A lightweight SQLite wrapper for Dart/Flutter with reactive queries, a type-safe schema builder, and zero native setup.

PulseDb.open() is pure Dart and works in any Dart project. Flutter-specific features (lifecycle mixin, openAsync()) are in a separate module.

The v1.0 backlog is complete — all 12 items are implemented. See BACKLOG.md for details.

Features

• Reactive streams — watch(table) and watchQuery(sql) emit fresh results on every insert/update/delete
• Paginated streams — watchCursor(table, {limit, offset}) for paginated reactive queries
• Isolate offloading — db.compute((db) => ...) runs queries on a separate isolate, keeping the UI thread free
• Bulk operations — insertAll, updateAll, upsert for batch writes
• Typed exceptions — Full PulseDbException hierarchy (PulseDbConstraintException, PulseDbSchemaException, etc.)
• Type-safe schema — TableDef, Col, and helpers like integer(), text(), real(), blob() with chainable modifiers (primaryKey(), required(), defaultTo(), autoIncrement())
• Typed repository — Repository<T> base class with insert, get, update, delete, deleteWhere, watch, watchWhere; plus MapRepository for map-based usage with no model class
• Lifecycle mixin — PulseDbMixin with initDb, observe(), autoObserve(), error recovery (initError, onError), and auto-dispose. ObservableList provides .isLoading, .isEmpty, .repo — no dbReady checks needed
• Auto schema sync — pass tables: to open() — auto-creates tables and adds new columns on schema changes. No manual migrations for table changes
• Migrations — versioned Migration list with up/down support, migrateTo(version), and previewMigrations() dry-run
• Concurrent access guard — automatic .db.lock file prevents double-opening from multiple instances
• No native setup — backed by sqlite3 v3 which bundles the native library via Dart hooks

Getting started

Add the dependency:

dependencies:
  pulse_db: ^1.0.0

No platform-specific configuration needed — sqlite3 v3 handles native library bundling automatically.

Usage

Schema definition

final todoTable = TableDef('todos', [
  integer('id').primaryKey().autoIncrement(),
  text('title').required(),
  text('note').defaultTo("''"),
  integer('priority').defaultTo('0'),
  integer('done').defaultTo('0'),
  text('created_at').defaultTo("(datetime('now'))"),
]);

Typed repository

No subclass needed — create Repository<T> directly:

final repo = Repository<Todo>(db, table: todoTable,
  fromRow: Todo.fromMap,
  toRow: (t) => t.toMap(),
);

repo.watch();              // Stream<List<Todo>>
repo.insert(todo);         // returns row id
repo.get(1);               // Todo?
repo.update({'done': 1}, where: 'id = ?', whereArgs: [1]);
repo.delete(1);

Or use MapRepository to skip the model class entirely:

final repo = MapRepository(db, todoTable);
repo.insert({'title': 'Learn', 'done': 0});
repo.watch();  // Stream<List<Map<String, dynamic>>>

With domain methods (optional)

Extend Repository<T> only when you need custom queries or domain logic:

class TodoRepository extends Repository<Todo> {
  TodoRepository(PulseDb db) : super(db, table: todoTable,
    fromRow: Todo.fromMap, toRow: (t) => t.toMap());

  Stream<List<Todo>> watchActive() => watchWhere('done = 0');
  void toggle(int id, bool done) =>
      update({'done': done ? 1 : 0}, where: 'id = ?', whereArgs: [id]);
  void remove(int id) => delete(id);
}

Using PulseDbMixin in a StatefulWidget

class _TodoPageState extends State<TodoPage> with PulseDbMixin {
  late final _todos = observe(Repository<Todo>(
    db, table: todoTable,
    fromRow: Todo.fromMap,
    toRow: (t) => t.toMap(),
  ));

  @override
  void initState() {
    super.initState();
    initDb(databaseName: 'todos.db', tables: [todoTable]);
  }

  @override
  Widget build(BuildContext context) {
    if (_todos.isLoading) {
      return const Center(child: CircularProgressIndicator());
    }
    if (_todos.isEmpty) return const Text('No items');
    final todos = _todos.value;
    // ... render list
  }
}

observe() returns an ObservableList<R> — a ValueNotifier subclass with .isLoading, .isEmpty, and .repo for insert/update/delete. It defers the watch subscription until the DB is ready, so isLoading stays true until the first data arrives.

For cases where the repository depends on PulseDb (which isn't ready yet at field init), use autoObserve() which defers creation:

late final _todos = autoObserve((db) => Repository<Todo>(db,
    table: todoTable, fromRow: Todo.fromMap, toRow: (t) => t.toMap()));

Map-based repository (no model class)

Skip the model class entirely with MapRepository — works directly with Map<String, dynamic>:

final repo = db.repository(todoTable);
// or: MapRepository(db, todoTable)

repo.insert({'title': 'Learn pulse_db', 'done': 0});
repo.watch().listen(print); // Stream<List<Map<String, dynamic>>>
repo.get(1);  // Map<String, dynamic>?
repo.delete(1);

Auto schema sync

Pass tables: to open() — tables are auto-created, and new columns are added automatically when the TableDef changes:

final db = PulseDb();
db.open(path: 'app.db', tables: [todoTable]);
// First run: CREATE TABLE
// After adding a column to todoTable: ALTER TABLE ADD COLUMN

No manual Migration tracking needed for schema evolution. Old migrations are still supported for data transformations.

Async opening

For convenience, openAsync() handles path resolution automatically:

final db = await openAsync(
  databaseName: 'app.db',
  tables: [todoTable],
);

Low-level API

final db = PulseDb();
db.open(path: 'app.db');

final id = db.insert('users', {'name': 'Alice'});
db.update('users', {'name': 'Bob'}, 'id = ?', [id]);
final rows = db.query('SELECT * FROM users');
db.delete('users', 'id = ?', [id]);

db.transaction(() {
  db.insert('users', {'name': 'Charlie'});
});

final stream = db.watch('users'); // reactive
db.close();

Paginated watch streams

Pagination via watchCursor — only re-queries the requested window:

final stream = db.watchCursor('users', limit: 20, offset: 40);

Works on Repository<T> too:

final stream = repo.watchCursor(limit: 10, offset: 0);

Isolate offloading

Offload heavy queries to a background isolate with compute():

final result = await db.compute((db) {
  return db.query('SELECT * FROM large_table WHERE ...');
});

Each call opens a fresh connection in a separate isolate and returns the result as a Future. Not supported on :memory: databases.

Repository<T> exposes the same API:

final result = await repo.compute((db) => db.query('SELECT COUNT(*) FROM ...'));

Bulk operations

Insert, update, or upsert multiple rows atomically:

db.insertAll('users', [
  {'name': 'Alice', 'age': 30},
  {'name': 'Bob', 'age': 25},
]);

db.updateAll('users', {'age': 0}, [1, 2, 3]); // reset ages for ids 1,2,3

db.upsert('users', {'id': 1, 'name': 'Alice'}, 'id'); // insert or replace

All available on Repository<T> and MapRepository too.

Migrations

Versioned migrations with up/down rollback:

final migrations = [
  Migration(version: 1, up: 'CREATE TABLE ...', down: 'DROP TABLE ...'),
  Migration(version: 2, up: 'ALTER TABLE ...', down: 'ALTER TABLE ...'),
];

db.open(path: 'app.db', migrations: migrations);

// Later: migrate up to v2 or down to v1
db.migrateTo(2, migrations); // forward
db.migrateTo(0, migrations); // roll back all

// Dry-run — see pending without applying
db.previewMigrations(migrations); // [Migration(1), Migration(2)]

Error recovery in the mixin

initDb() captures failures gracefully — the widget won't crash:

await initDb(databaseName: 'app.db', tables: [todoTable],
  onError: (error, stack) => log(error),
);
if (initError != null) return Text('Error: $initError');

Exception handling

All operations throw typed exceptions from the PulseDbException hierarchy, making it easy to catch specific database errors:

import 'package:pulse_db/pulse_db.dart';

try {
  db.insert('users', {'id': 1, 'name': 'Alice'});
  db.insert('users', {'id': 1, 'name': 'Bob'}); // Same ID
} on PulseDbConstraintException catch (e) {
  print('Constraint violation: $e');
} on PulseDbSchemaException catch (e) {
  print('Schema error (e.g. missing table): $e');
} on PulseDbTransactionException catch (e) {
  print('Transaction failed: $e');
} on PulseDbException catch (e) {
  print('Other database error: $e');
}

Additional information

• See the /example directory for a complete todo-list app
• File issues at github.com/carllosnc/pulse_db
• Pull requests welcome

---

Carlos Costa @ 2026