58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""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]
|