Examples

Every example on this page is a complete runnable program executed by the test suite on every change; they cannot go stale.

Conflict is data, not an exception: detect it, resolve it, lose nothing

"""Conflict is data, not an exception: detect it, resolve it, lose nothing."""

import tempfile
from pathlib import Path

import potluckdb

with tempfile.TemporaryDirectory() as scratch:
    root = Path(scratch)
    remote = potluckdb.Remote.from_directory(root / "remote")
    manual = potluckdb.OpenOptions(sync_policy=potluckdb.SyncPolicy.manual())

    writer = potluckdb.open(root / "writer", remote, options=manual)
    writer.execute("CREATE TABLE item(id INTEGER PRIMARY KEY, value TEXT)")
    writer.sync()

    rival = potluckdb.open(root / "rival", remote, options=manual)
    rival.execute("INSERT INTO item(id, value) VALUES(1, 'from rival')")
    assert type(rival.sync()).__name__ == "Pushed"

    # The writer races the same head with its own local work.
    writer.execute("INSERT INTO item(id, value) VALUES(2, 'from writer')")
    conflict = writer.sync()
    assert type(conflict).__name__ == "Conflict"

    resolution = writer.resolve_conflict()
    assert resolution.revision is not None
    outcome = writer.sync()
    assert type(outcome).__name__ in {"Pushed", "UpToDate"}

    # Both sides' rows survive: the winner was adopted and the safe local
    # statement replayed on top of it.
    rows = writer.query("SELECT id, value FROM item ORDER BY id")
    assert list(rows) == [(1, "from rival"), (2, "from writer")]

    writer.close()
    rival.close()
    remote.close()

print("conflict_resolution: ok")

Local use needs no configuration: open, write, read, close

"""Local use needs no configuration: open, write, read, close."""

import tempfile
from pathlib import Path

import potluckdb

with tempfile.TemporaryDirectory() as scratch:
    db = potluckdb.open(Path(scratch) / "app.potluckdb")
    db.execute("CREATE TABLE note(id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
    result = db.execute("INSERT INTO note(body) VALUES(?)", ("hello",))
    assert result.rows_affected == 1
    assert result.last_insert_id == 1

    rows = db.query("SELECT id, body FROM note ORDER BY id")
    assert list(rows) == [(1, "hello")]
    db.close()
    db.close()  # close is idempotent

print("quickstart_local: ok")

"""Two devices share one remote: write on one, sync, read on the other."""

import tempfile
from pathlib import Path

import potluckdb

with tempfile.TemporaryDirectory() as scratch:
    root = Path(scratch)
    remote = potluckdb.Remote.from_directory(root / "remote")
    manual = potluckdb.OpenOptions(sync_policy=potluckdb.SyncPolicy.manual())

    laptop = potluckdb.open(root / "laptop", remote, options=manual)
    laptop.execute("CREATE TABLE task(id INTEGER PRIMARY KEY, title TEXT)")
    laptop.execute("INSERT INTO task(title) VALUES(?)", ("ship the release",))
    outcome = laptop.sync()
    assert type(outcome).__name__ == "Pushed"

    phone = potluckdb.open(root / "phone", remote, options=manual)
    rows = phone.query("SELECT title FROM task")
    assert list(rows) == [("ship the release",)]

    # An unchanged head is reported as exactly that.
    assert type(phone.sync()).__name__ == "UpToDate"

    laptop.close()
    phone.close()
    remote.close()

print("remote_sync: ok")

Managed transactions: commit on success, rollback on failure

"""Managed transactions: commit on success, rollback on failure."""

import tempfile
from pathlib import Path

import potluckdb

with tempfile.TemporaryDirectory() as scratch:
    db = potluckdb.open(Path(scratch) / "app.potluckdb")
    db.execute("CREATE TABLE ledger(id INTEGER PRIMARY KEY, amount INTEGER)")

    with db.transaction() as transaction:
        transaction.execute("INSERT INTO ledger(amount) VALUES(?)", (100,))
        transaction.execute("INSERT INTO ledger(amount) VALUES(?)", (-40,))
        (total,) = transaction.query("SELECT SUM(amount) FROM ledger")[0]
        assert total == 60
    # The scope completed, so both rows committed.
    assert db.query("SELECT COUNT(*) FROM ledger")[0][0] == 2

    try:
        with db.transaction() as transaction:
            transaction.execute("INSERT INTO ledger(amount) VALUES(?)", (999,))
            raise RuntimeError("abort this batch")
    except RuntimeError:
        pass
    # The failed scope rolled back: still two rows.
    assert db.query("SELECT COUNT(*) FROM ledger")[0][0] == 2

    db.close()

print("transactions: ok")