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)
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_allto add the file to a database and use the ROWID as primary key by settingpk="id".However, I want to use ULID as primary keys instead OR create a
ulidcolumn that will be populated alongside theinsert_alloperation.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_functionsmethod have not been successful.