e7099d250b
* Working on making export threadsafe, #999 * Working on making export threadsafe, #999 * refactor for concurrent export, #999 * Fixed race condition in ExportRecord context manager
62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""sqlite utils for use by osxphotos"""
|
|
|
|
import logging
|
|
import pathlib
|
|
import sqlite3
|
|
from typing import List, Tuple
|
|
|
|
from ._constants import SQLITE_CHECK_SAME_THREAD
|
|
|
|
logger = logging.getLogger("osxphotos")
|
|
|
|
|
|
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,
|
|
check_same_thread=SQLITE_CHECK_SAME_THREAD,
|
|
)
|
|
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"""
|
|
|
|
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 as e:
|
|
logger.debug(f"sqlite_db_is_locked: {e}")
|
|
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]
|