Implemented #731, export_id in report database

osxphotos/sqlite_utils.py

@@ -0,0 +1,57 @@
+"""sqlite utils for use by osxphotos"""
+
+import os.path
+import pathlib
+import sqlite3
+from typing import List, Tuple
+
+
+def sqlite_open_ro(dbname: str) -> Tuple[sqlite3.Connection, sqlite3.Cursor]:
+    """opens sqlite file dbname in read-only mode
+    returns tuple of (connection, cursor)"""
+    try:
+        dbpath = pathlib.Path(dbname).resolve()
+        conn = sqlite3.connect(f"{dbpath.as_uri()}?mode=ro", timeout=1, uri=True)
+        c = conn.cursor()
+    except sqlite3.Error as e:
+        raise sqlite3.Error(
+            f"An error occurred opening sqlite file: {e} {dbname}"
+        ) from e
+    return (conn, c)
+
+
+def sqlite_db_is_locked(dbname):
+    """check to see if a sqlite3 db is locked
+    returns True if database is locked, otherwise False
+    dbname: name of database to test"""
+
+    # first, check to see if lock file exists, if so, assume the file is locked
+    lock_name = f"{dbname}.lock"
+    if os.path.exists(lock_name):
+        return True
+
+    # no lock file so try to read from the database to see if it's locked
+    locked = None
+    try:
+        (conn, c) = sqlite_open_ro(dbname)
+        c.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;")
+        conn.close()
+        locked = False
+    except Exception:
+        locked = True
+
+    return locked
+
+
+def sqlite_tables(conn: sqlite3.Connection) -> List[str]:
+    """Returns list of tables found in sqlite db"""
+    results = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table';"
+    ).fetchall()
+    return [row[0] for row in results]
+
+
+def sqlite_columns(conn: sqlite3.Connection, table: str) -> List[str]:
+    """Returns list of column names found in table in sqlite database"""
+    results = conn.execute(f"PRAGMA table_info({table});")
+    return [row[1] for row in results]