refactor: migrate from PysonDB to SQLite for database management

Signed-off-by: Jos Verlinde <Jos.Verlinde@microsoft.com>

pyproject.toml

@@ -68,7 +68,6 @@ pipx = "^1.1.0"
 pygithub = "^2.1.1"
 pyright = ">=1.1.341"
 pyserial = ">=3.5"
-pysondb-v2 = "^2.1.0"
 pypi-simple = "^1.0.0"
 rich-click = "^1.8.3"
 tabulate = "^0.9.0"

src/stubber/publish/database.py

@@ -1,26 +1,59 @@
-"""basic interface to the json database"""
+"""Interface to the database which stores the package information"""
 
 import sqlite3
 from pathlib import Path
-from typing import Union
 
-from pysondb import PysonDB
 
-
-def get_database(publish_path: Path, production: bool = False) -> sqlite3.Connection:
+def get_database(db_path: Path, production: bool = False) -> sqlite3.Connection:
     """
     Open the sqlite database at the given path.
 
     The database should be located in a subfolder `/data` of the root path.
     The database name is determined by the production flag as `all_packages[_test].db`
     """
+    db_path = Path(db_path)
+    if db_path.stem == "publish":
+        db_path = db_path / ".." / "data"  # go up one level to find the database
 
-    publish_path = Path(publish_path)
-    db_path = publish_path / f"all_packages{'' if production else '_test'}.db"
-    if not db_path.exists():
-        raise FileNotFoundError("Database file not found")
+    db_path = db_path / f"all_packages{'' if production else '_test'}.db"
+    if db_path.exists():
+        conn = sqlite3.connect(db_path)
 
-    conn = sqlite3.connect(db_path)
+    else:
+        print(FileNotFoundError(f"Database file not found in path: {db_path}"))
+        conn = create_database(db_path)
+
+    conn.row_factory = sqlite3.Row  # return rows as dicts
     return conn
 
 
+def create_database(db_path: Path) -> sqlite3.Connection:
+    """
+    Create a new database at the given path.
+
+    The database should be located in a subfolder `/data` of the root path.
+    """
+    db_path = Path(db_path)
+    if not db_path.parent.exists():
+        db_path.parent.mkdir(parents=True)
+    conn = sqlite3.connect(db_path)
+    SCHEMA = """
+    CREATE TABLE "packages" (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT,
+            description TEXT,
+            mpy_version TEXT,
+            pkg_version TEXT,
+            publish BOOLEAN,
+            stub_sources TEXT, -- json string
+            path TEXT,
+            hash TEXT,
+            stub_hash TEXT,
+            port TEXT DEFAULT "",
+            board TEXT DEFAULT "",
+            variant TEXT DEFAULT ""
+        )
+    """
+    conn.execute(SCHEMA)
+    conn.commit
+    return conn

src/stubber/publish/package.py

@@ -10,8 +10,7 @@
 
 from mpflash.logger import log
 from mpflash.versions import clean_version
-from packaging.version import parse
-from pysondb import PysonDB
+
 
 from stubber.publish.defaults import GENERIC, GENERIC_L, default_board
 from stubber.publish.enums import StubSource
@@ -78,18 +77,16 @@ def get_package_info(
     """
     # find in the database
 
-    SQL_Q = f"""
-        SELECT * 
-            FROM packages 
-            WHERE name = '{pkg_name}' AND mpy_version LIKE '{mpy_version}%'
-            ORDER BY pkg_version DESC
-    """
-    log.trace(f"SQL Query: {SQL_Q}")
     cursor = db_conn.cursor()
-    cursor.execute(SQL_Q)
-    cursor.row_factory = sqlite3.Row # to get dict like access
+    cursor.execute(
+        """
+        SELECT * FROM packages 
+        WHERE name = ? AND mpy_version LIKE ?
+        ORDER BY pkg_version DESC
+    """,
+        (pkg_name, f"{mpy_version}%"),
+    )
     packages = cursor.fetchall()
-    log.debug(f"Found {len(packages)} packages for {pkg_name} == {mpy_version}")
 
     if len(packages) > 0:
         pkg_from_db = dict(packages[-1])

src/stubber/publish/stubpackage.py

@@ -25,7 +25,6 @@
 from mpflash.logger import log
 from mpflash.versions import SET_PREVIEW, V_PREVIEW, clean_version
 from packaging.version import Version, parse
-from pysondb import PysonDB
 
 from stubber.publish.bump import bump_version
 from stubber.publish.defaults import GENERIC_U, default_board
@@ -216,6 +215,7 @@ def __init__(
         mpy_version: str = "0.0.1",
         port: str,
         board: str = GENERIC_U,
+        variant: Optional[str] = None,
         description: str = "MicroPython stubs",
         stubs: Optional[StubSources] = None,
         # json_data: Optional[Dict[str, Any]] = None,
@@ -226,6 +226,7 @@ def __init__(
         self.mpy_version = mpy_version
         self.port = port
         self.board = board
+        self.variant = variant or ""
         self.description = description
         self.stub_sources = stubs or []
         self.hash = None  # intial hash
@@ -405,6 +406,9 @@ def to_dict(self) -> dict:
             "description": self.description,
             "hash": self.hash,
             "stub_hash": self.stub_hash,
+            "port": self.port,
+            "board": self.board,
+            "variant": "",  # TODO: get the variant
         }
 
     def from_dict(self, json_data: Dict) -> None:
@@ -416,6 +420,10 @@ def from_dict(self, json_data: Dict) -> None:
         self._publish = json_data["publish"]
         self.hash = json_data["hash"]
         self.stub_hash = json_data["stub_hash"]
+        self.port = json_data["port"]
+        self.board = json_data["board"]
+        self.variant = json_data["variant"]
+
         # create folder
         if not self.package_path.exists():
             self.package_path.mkdir(parents=True, exist_ok=True)
@@ -1030,14 +1038,30 @@ def publish_distribution(
             log.warning(f"{self.package_name}: Dry run, not saving to database")
         else:
             cursor = db_conn.cursor()
-            variant = ""  # TODO: get the variant
-            row = f"""
-            INSERT INTO packages (id, name, description, mpy_version, pkg_version, publish, stub_sources, path, hash, stub_hash, port, board, variant)
-            VALUES ({key}, '{self.package_name}', '{self.description}', '{self.mpy_version}', 
-                '{self.pkg_version}', {self._publish}, '{json.dumps(self.stub_sources)}', '{self.package_path}, '{variant}', 
-                '{self.hash}', '{self.stub_hash} ', '{self.port}', '{self.board}');
-            """
-            cursor.execute(row)
+
+            d = self.to_dict()
+
+            cursor.execute(
+                """
+                INSERT INTO packages (name, description, mpy_version, pkg_version, publish, stub_sources, path, hash, stub_hash, port, board, variant)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    d["name"],
+                    d["description"],
+                    d["mpy_version"],
+                    d["pkg_version"],
+                    d["publish"],
+                    json.dumps(d["stub_sources"]),
+                    d["path"],
+                    d["hash"],
+                    d["stub_hash"],
+                    d["port"],
+                    d["board"],
+                    d["variant"],
+                ),
+            )
+
             # # get the package state and add it to the database
             # db_conn.add(self.to_dict())
             db_conn.commit()

tests/publish/conftest.py

@@ -1,17 +1,18 @@
 """pytest fixtures for publish tests"""
 
 import shutil
+import sqlite3
 from pathlib import Path
 
 import pytest
-from pysondb import PysonDB
 from pytest_mock import MockerFixture
-
 from stubber.publish.package import create_package
 
 from .fakeconfig import FakeConfig
+
 pytestmark = [pytest.mark.stubber]
 
+
 @pytest.fixture
 def fake_package(request, mocker: MockerFixture, tmp_path: Path, pytestconfig: pytest.Config):
     """\
@@ -32,26 +33,28 @@ def fake_package(request, mocker: MockerFixture, tmp_path: Path, pytestconfig: p
         # use default version
         version = "1.19.1"
     pkg = create_package("micropython-fake-stubs", mpy_version=version, port="esp32")
-    pkg._publish = False # type: ignore
+    pkg._publish = False  # type: ignore
     pkg.create_license()
     pkg.create_readme()
     yield pkg
 
 
 @pytest.fixture
-def temp_db(
+def temp_db_conn(
     pytestconfig: pytest.Config,
     tmp_path: Path,
 ):
     """"""
-    db_src = pytestconfig.rootpath / "tests/publish/data/package_data_test.jsondb"
-    db_path = tmp_path / "package_data_test.jsondb"
+    db_src = pytestconfig.rootpath / "tests/publish/data/all_packages_test.db"
+    db_path = tmp_path / "all_packages_test.db"
     # copy file to temp location
     shutil.copy(db_src, db_path)
-    db = PysonDB(db_path.as_posix())
 
-    def fake_commit():
+    db_conn = sqlite3.connect(db_path)
+    db_conn.row_factory = sqlite3.Row # return rows as dicts
+    yield db_conn
+    try:
+        db_conn.close()
+        db_path.unlink()
+    except Exception as e:
         pass
-
-    db.commit = fake_commit
-    yield db