File size: 1,909 Bytes
6791083 4dc29ae 6791083 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
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 |