Get user profile snapshot from N days ago
(user_id: str, days_ago: int)
| 1199 | |
| 1200 | |
| 1201 | def get_profile_snapshot(user_id: str, days_ago: int) -> Optional[Dict]: |
| 1202 | """Get user profile snapshot from N days ago""" |
| 1203 | conn = get_connection() |
| 1204 | cursor = conn.cursor() |
| 1205 | |
| 1206 | target_date = datetime.now() - timedelta(days=days_ago) |
| 1207 | target_date_str = target_date.date().isoformat() |
| 1208 | |
| 1209 | cursor.execute(""" |
| 1210 | SELECT profile_json, updated_at FROM profiles |
| 1211 | WHERE user_id = ? AND date(updated_at) <= ? |
| 1212 | ORDER BY updated_at DESC |
| 1213 | LIMIT 1 |
| 1214 | """, (user_id, target_date_str)) |
| 1215 | |
| 1216 | row = cursor.fetchone() |
| 1217 | conn.close() |
| 1218 | |
| 1219 | if row: |
| 1220 | return json.loads(row["profile_json"]) |
| 1221 | return None |
| 1222 | |
| 1223 | |
| 1224 | def get_direction_changes(user_id: str, days: int = 7) -> List[Dict]: |
no test coverage detected