MinCPionS / sql /sql_utils.py
KoRiF
improve data lifecycle
6791083
import os
import sqlite3
from datetime import datetime, timedelta
def load_sql_query(filename: str) -> str:
"""Load SQL query from file"""
with open(filename, 'r') as f:
return f.read()
SQL_SELECT_METADATA = "sql/select_metadata.sql"
SQL_UPDATE_METADATA = "sql/update_metadata.sql"
SQL_CREATE_METADATA = "sql/create_metadata.sql"
def create_metadata_table(db_path: str):
"""Create metadata table if it doesn't exist"""
query = load_sql_query(SQL_CREATE_METADATA)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(query)
conn.commit()
def update_db_timestamp(db_path: str):
"""Update last database fetch timestamp"""
query = load_sql_query(SQL_UPDATE_METADATA)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(query, ("last_update", datetime.now().isoformat()))
conn.commit()
def is_database_outdated(db_path: str, age_threshold_hours: int = 24) -> bool:
"""
Check if database needs update
Returns True if:
- Database file doesn't exist
- No last_update metadata
- Last update was more than age_threshold_hours ago
"""
if not os.path.exists(db_path):
return True
query = load_sql_query(SQL_SELECT_METADATA)
try:
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(query, ("last_update",))
result = cursor.fetchone()
if not result:
return True
last_update = datetime.fromisoformat(result[0])
age_threshold = timedelta(hours=age_threshold_hours)
return datetime.now() - last_update > age_threshold
except (sqlite3.Error, ValueError) as e:
print(f"Error checking database age: {e}")
return True