Zalgorithm

Create and populate an SQLite database with Python

GitHub:

Given a JSON file with the following form, create and populate an SQLite database:

{
  "tmux_version": "3.7c",
  "source": "tmux(1), DEFAULT KEY BINDINGS",
  "prefix": "C-b",
  "bindings": [
    {
      "key": "C-b",
      "description": "Send the prefix key (C-b) through to the application."
    },
    {
      "key": "C-o",
      "description": "Rotate the panes in the current window forwards."
    },
    {
      "key": "C-z",
      "description": "Suspend the tmux client."
    }
}

Python:

"""Populate SQLite with the default key bindings from the tmux manual."""

import argparse
import json
import sqlite3
from pathlib import Path

DATA_PATH = Path(__file__).parent / "data" / "tmux_key_bindings.json"


def populate(database: Path) -> int:
    """Insert or update the bundled bindings in a single transaction."""
    data = json.loads(DATA_PATH.read_text(encoding="utf-8"))
    connection = sqlite3.connect(database)
    try:
        with connection:
            connection.execute("""
                CREATE TABLE IF NOT EXISTS key_bindings (
                    id INTEGER PRIMARY KEY,
                    key TEXT NOT NULL UNIQUE,
                    description TEXT NOT NULL,
                    prefix TEXT NOT NULL,
                    tmux_version TEXT NOT NULL,
                    source TEXT NOT NULL
                )
            """)
            connection.executemany(
                """
                INSERT INTO key_bindings
                    (key, description, prefix, tmux_version, source)
                VALUES (?, ?, ?, ?, ?)
                ON CONFLICT(key) DO UPDATE SET
                    description = excluded.description,
                    prefix = excluded.prefix,
                    tmux_version = excluded.tmux_version,
                    source = excluded.source
                """,
                [
                    (
                        binding["key"],
                        binding["description"],
                        data["prefix"],
                        data["tmux_version"],
                        data["source"],
                    )
                    for binding in data["bindings"]
                ],
            )
    finally:
        connection.close()
    return len(data["bindings"])


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "database",
        nargs="?",
        type=Path,
        default=Path("tmux_key_bindings.sqlite3"),
        help="SQLite file to populate (default: tmux_key_bindings.sqlite3)",
    )
    args = parser.parse_args()
    count = populate(args.database)
    print(f"Loaded {count} key binding entries into {args.database}.")


if __name__ == "__main__":
    main()

Run with python populate.py <optional_db_name>, or for the project’s virtual env, run with uv run populate.py <optional_db_name>.

The Python sqlite3 module #

Documentation: https://docs.python.org/3/library/sqlite3.html

The Python docs example uses a Connection object and a Cursor object:

import sqlite3
con = sqlite3.connect("tutorial.db")
cur = con.cursor()

But Python’s Connection object provides methods that create a cursor internally. These two forms are equivalent:

cursor = connection.cursor()
cursor.execute("SELECT * FROM key_bindings")

cursor = connection.execute("SELECT * FROM key_bindings")

Closing database connections #

The connection opened with connection = sqlite3.connect(database) needs to be closed. In the current code, try/finally statements are used:

    try:
        with connection:
            connection.execute("""
			# ...
    finally:
        connection.close()

contextlib.closing #

A possibly cleaner alternative is to use contextlib.closing. It guarantees that connection.close() gets called when the block exits, including when exceptions offur. Uset it alongside the connection’s own context manager:

from contextlib import closing

with closing(sqlite3.connect(database)) as connection:
  with connection:
	  connection.execute(...)
	  connection.executemany(...)

Each bock has a separate job:

SQL upsert #

Related to:

"""
INSERT INTO key_bindings
(key, description, prefix, tmux_version, source)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(key) DO UPDATE SET
description = excluded.description,
prefix = excluded.prefix,
tmux_version = excluded.tmux_version,
source = excluded.source
""",

ON CONFLICT(key) DO UPDATE SET makes the insert an upsert. The table defines key as UNIQUE—two rows can’t have the same key. ON CONFLICT(key) DO UPDATE SET means “if this insert would duplicate an existing key, update the existing row.”

excluded (e.g. excluded.prefix) refers to the proposed new row—the value that is being inserted. So prefix = excluded.prefix means replace the existing row’s prefix with the new prefix.

Tags: