Use sqlite3.Row (#189)

Author: mxr

sqlite_export_for_ynab/_main.py

@@ -106,7 +106,7 @@ async def sync(token: str, db: Path, full_refresh: bool) -> None:
         db.parent.mkdir(parents=True, exist_ok=True)
 
     with sqlite3.connect(db) as con:
-        con.row_factory = _row_factory
+        con.row_factory = sqlite3.Row
         cur = con.cursor()
 
         if full_refresh:
@@ -185,10 +185,6 @@ async def sync(token: str, db: Path, full_refresh: bool) -> None:
             print("Done")
 
 
-def _row_factory(c: sqlite3.Cursor, row: tuple[Any, ...]) -> dict[str, Any]:
-    return {d[0]: r for d, r in zip(c.description, row, strict=True)}
-
-
 def contents(filename: str) -> str:
     return (resources.files(ddl) / filename).read_text()
 

testing/fixtures.py

@@ -8,7 +8,6 @@
 import pytest
 from aioresponses import aioresponses
 
-from sqlite_export_for_ynab._main import _row_factory
 from sqlite_export_for_ynab._main import contents
 
 PLAN_ID_1 = str(uuid4())
@@ -326,7 +325,7 @@
 @pytest.fixture
 def cur():
     with sqlite3.connect(":memory:") as con:
-        con.row_factory = _row_factory
+        con.row_factory = sqlite3.Row
         cursor = con.cursor()
         cursor.executescript(contents("create-relations.sql"))
         yield cursor
@@ -338,8 +337,8 @@ def mock_aioresponses():
         yield m
 
 
-def strip_nones(d: dict[str, Any]) -> dict[str, Any]:
-    return {k: v for k, v in d.items() if v is not None}
+def strip_nones(d: sqlite3.Row) -> dict[str, Any]:
+    return {k: v for k, v in dict(d).items() if v is not None}
 
 
 TOKEN = f"token-{uuid4()}"

tests/_main_test.py

@@ -106,7 +106,7 @@ def test_get_last_knowledge_of_server(cur):
 def test_insert_plans(cur):
     insert_plans(cur, PLANS, LKOS)
     cur.execute("SELECT * FROM plans ORDER BY name")
-    assert cur.fetchall() == [
+    assert [strip_nones(d) for d in cur.fetchall()] == [
         {
             "id": PLAN_ID_1,
             "name": PLANS[0]["name"],
@@ -170,7 +170,7 @@ def test_insert_accounts(cur):
     ]
 
     cur.execute("SELECT * FROM account_periodic_values ORDER BY name")
-    assert cur.fetchall() == [
+    assert [strip_nones(d) for d in cur.fetchall()] == [
         {
             "account_id": ACCOUNT_ID_1,
             "plan_id": PLAN_ID_1,