Coverage for health / db / schema.py: 0%

1""" 

2SQLite database schema for Garmin Health Sync system. 

3 

4Defines tables for sync tracking and data indexing. 

5""" 

6 

7import sqlite3 

8from pathlib import Path 

9from typing import Optional 

10 

11from health.config import DB_PATH 

12from health.utils.logging_config import setup_logger 

13 

14logger = setup_logger(__name__) 

15 

16 

17# SQL statements for creating tables 

18CREATE_SYNC_RECORDS_TABLE = """ 

19CREATE TABLE IF NOT EXISTS sync_records ( 

20 id INTEGER PRIMARY KEY AUTOINCREMENT, 

21 data_type TEXT NOT NULL, 

22 start_date TEXT NOT NULL, 

23 end_date TEXT NOT NULL, 

24 status TEXT NOT NULL, -- 'success', 'failed', 'partial' 

25 records_synced INTEGER DEFAULT 0, 

26 error_message TEXT, 

27 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 

28); 

29""" 

30 

31CREATE_LAST_SYNC_STATE_TABLE = """ 

32CREATE TABLE IF NOT EXISTS last_sync_state ( 

33 data_type TEXT PRIMARY KEY, 

34 last_sync_date TEXT NOT NULL, 

35 total_records INTEGER DEFAULT 0, 

36 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 

37); 

38""" 

39 

40CREATE_DAILY_METRICS_INDEX_TABLE = """ 

41CREATE TABLE IF NOT EXISTS daily_metrics_index ( 

42 id INTEGER PRIMARY KEY AUTOINCREMENT, 

43 metric_type TEXT NOT NULL, 

44 date TEXT NOT NULL, 

45 file_path TEXT NOT NULL, 

46 has_data BOOLEAN DEFAULT 1, 

47 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 

48 UNIQUE(metric_type, date) 

49); 

50""" 

51 

52CREATE_ACTIVITY_INDEX_TABLE = """ 

53CREATE TABLE IF NOT EXISTS activity_index ( 

54 id INTEGER PRIMARY KEY AUTOINCREMENT, 

55 activity_id TEXT UNIQUE NOT NULL, 

56 activity_type TEXT, 

57 date TEXT NOT NULL, 

58 duration_seconds INTEGER, 

59 distance_meters REAL, 

60 file_path TEXT NOT NULL, 

61 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 

62); 

63""" 

64 

65# Indexes for performance 

66CREATE_INDEXES = [ 

67 "CREATE INDEX IF NOT EXISTS idx_sync_records_type_date ON sync_records(data_type, start_date);", 

68 "CREATE INDEX IF NOT EXISTS idx_daily_metrics_type_date ON daily_metrics_index(metric_type, date);", 

69 "CREATE INDEX IF NOT EXISTS idx_activity_date ON activity_index(date);", 

70 "CREATE INDEX IF NOT EXISTS idx_activity_type ON activity_index(activity_type);", 

71] 

72 

73 

74def get_connection(db_path: Optional[Path] = None) -> sqlite3.Connection: 

75 """Get a database connection. 

76 

77 Args: 

78 db_path: Optional path to database file (defaults to config.DB_PATH) 

79 

80 Returns: 

81 SQLite connection object 

82 """ 

83 path = db_path or DB_PATH 

84 conn = sqlite3.connect(str(path)) 

85 conn.row_factory = sqlite3.Row # Enable column access by name 

86 return conn 

87 

88 

89def init_database(db_path: Optional[Path] = None) -> None: 

90 """Initialize database schema. 

91 

92 Creates all necessary tables and indexes if they don't exist. 

93 

94 Args: 

95 db_path: Optional path to database file (defaults to config.DB_PATH) 

96 """ 

97 path = db_path or DB_PATH 

98 logger.info(f"Initializing database at {path}") 

99 

100 with get_connection(path) as conn: 

101 cursor = conn.cursor() 

102 

103 # Create tables 

104 cursor.execute(CREATE_SYNC_RECORDS_TABLE) 

105 cursor.execute(CREATE_LAST_SYNC_STATE_TABLE) 

106 cursor.execute(CREATE_DAILY_METRICS_INDEX_TABLE) 

107 cursor.execute(CREATE_ACTIVITY_INDEX_TABLE) 

108 

109 # Create indexes 

110 for index_sql in CREATE_INDEXES: 

111 cursor.execute(index_sql) 

112 

113 conn.commit() 

114 

115 logger.info("Database initialization complete") 

116 

117 

118def reset_database(db_path: Optional[Path] = None) -> None: 

119 """Reset database by dropping all tables and reinitializing. 

120 

121 WARNING: This will delete all sync history and indexes! 

122 

123 Args: 

124 db_path: Optional path to database file (defaults to config.DB_PATH) 

125 """ 

126 path = db_path or DB_PATH 

127 logger.warning(f"Resetting database at {path}") 

128 

129 with get_connection(path) as conn: 

130 cursor = conn.cursor() 

131 

132 # Drop all tables 

133 cursor.execute("DROP TABLE IF EXISTS sync_records") 

134 cursor.execute("DROP TABLE IF EXISTS last_sync_state") 

135 cursor.execute("DROP TABLE IF EXISTS daily_metrics_index") 

136 cursor.execute("DROP TABLE IF EXISTS activity_index") 

137 

138 conn.commit() 

139 

140 # Reinitialize 

141 init_database(path) 

142 logger.info("Database reset complete") 

143 

144 

145if __name__ == "__main__": 

146 # Allow running as script to initialize database 

147 init_database() 

148 print(f"Database initialized at {DB_PATH}")