simonw / simonw/sqlite-utils

Auto-generate ULID as Primary Keys on Bulk Insert

Open
#728 2 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
2.2k
Forks
172
Avg merge
9m
Merged PRs (30d)
1

Description

I want to add a JSON file from an API into SQLite database using Python. The endpoints give varied nested JSON.
I was able to flatten the nested responses.
I was able to use insert_all to add the file to a database and use the ROWID as primary key by setting pk="id".
However, I want to use ULID as primary keys instead OR create a ulid column that will be populated alongside the insert_all operation.

I could do something similar with sqlite3 module by registering a Python function and prepending SQL query before the final query.

However, all attempts to use the register_functions method have not been successful.


#!/usr/bin/env python3
from __future__ import annotations

import json
import orjson
import sqlite3
import sqlite_ulid
import sqlite_utils
import sys

from pathlib import Path
from rich import print as rprint
from ulid import ULID

if sys.version_info < (3, 9):
    from typing_extensions import Optional, Union
else:
    from typing import Optional, Union

# Query to setup the database for multi-threaded access
pragma_setup_multi_query = """
    PRAGMA journal_mode=WAL;
    PRAGMA busy_timeout=5000;
    PRAGMA mmap_size=268435456; -- 256 * 1024 * 1024
    PRAGMA cache_size=-64000;
    PRAGMA synchronous=NORMAL;
    PRAGMA foreign_keys=ON;
    PRAGMA enable_fts5;
    PRAGMA enable_extension = regexp;
"""

# Create a ULID function.
def ulid_func():
    return str(ULID())

def json_2_sqlite(
    path_to_json: Union[Path, str],
    path_to_sqlite: Optional[Union[Path, str]] = None,
    table_name: str = "data",
    messages: bool = True,
):
    path_to_json = Path(path_to_json).expanduser().absolute()
    try:
        with open(path_to_json, "r") as fin:
            data_fetched = orjson.loads(fin.read())
    except orjson.JSONDecodeError:
        with open(path_to_json, "r") as fin:
            data_fetched = json.load(fin)

    data_fetched_flatten = data_fetched
    
    if data_fetched_flatten is None:
        raise ValueError("No data fetched")

    if messages:
        rprint(f"Loaded {len(data_fetched_flatten)} records")
        rprint(f"Last record:\n{data_fetched_flatten[-1]}")

    if not path_to_sqlite:
        path_to_sqlite = Path("./test-code.sqlite")
    path_to_sqlite = Path(path_to_sqlite).expanduser().absolute()
    
    conn = sqlite3.connect(path_to_sqlite)
    conn.enable_load_extension(True)
    sqlite_ulid.load(conn)
    conn.create_function("ulid", 0, ulid_func)
    cursor = conn.cursor()
    cursor.executescript(pragma_setup_multi_query)
    conn.commit()
    conn.close()
    if messages:
        rprint(f"Database created at {path_to_sqlite}")

    db = sqlite_utils.Database(path_to_sqlite)
    db.register_function(ulid_func)

    db[table_name].insert_all(
        records=data_fetched_flatten,
        alter=True,
        pk="id"
    ).add_missing_columns([{"ulid": "", "created_at": "", "updated_at": ""}],) # using `str` created `BLOB` instead of `TEXT` columns
    
    # TODO: Prior Attempt
    db[table_name].insert_all(
        records=data_fetched_flatten,
        alter=True,
        pk=ulid_func #TypeError: object of type 'function' has no len()
    )
    
    # TODO: Prior Attempt
    db[table_name].insert_all(
        records=data_fetched_flatten,
        alter=True,
        pk=ulid_func() #Creates a column with a ULID then fills the rows with ROWID
    )

    db.close()
    

    if messages:
        rprint(f"Database updated at {path_to_sqlite}")


if __name__ == "__main__":
    json_2_sqlite(path_to_json="./test-code-launches-tiny.json", table_name="launches", messages=True)



Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the sqlite-utils Database.register_function and Table.insert_all entry points used in the example, then review how add_missing_columns handles the requested ulid column. Done means bulk insertion can generate a distinct ULID for each record, either as the primary key or in a populated ulid column, with the behavior demonstrated by a test.

Written by the indexing model from the issue text.

Assessment

Tech stack
python, sqlite
Domain
database
Issue type
Feature
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.