ryan
da7ed41938
- Implement lazy MongoDB client initialization in backend/db/mongo_client.py. - Create Gunicorn configuration to ensure MongoDB client is initialized per worker. - Refactor database access layer to support MongoDB with a new MongoLockedTable adapter. - Add migration script to transfer existing TinyDB data to MongoDB, preserving idempotency. - Update tracking event handling to ensure deterministic ordering with a monotonic sequence. - Modify tests to use mongomock for MongoDB integration and ensure existing tests pass. - Add integration test script to run tests against a local MongoDB Docker container. - Document environment variables and migration process in specs/feat-database-migration.md.
140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
"""Helper functions for tracking events database operations."""
|
|
import itertools
|
|
import logging
|
|
from typing import Optional, List
|
|
from tinydb import Query
|
|
from db.db import tracking_events_db
|
|
from models.tracking_event import TrackingEvent, EntityType, ActionType
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Monotonic sequence used as a deterministic tiebreaker when tracking events
|
|
# share the same ``occurred_at``/``created_at`` timestamps (common in tests).
|
|
_tracking_event_seq = itertools.count()
|
|
|
|
|
|
def insert_tracking_event(event: TrackingEvent) -> str:
|
|
"""
|
|
Insert a tracking event into the database.
|
|
|
|
Args:
|
|
event: TrackingEvent instance to insert
|
|
|
|
Returns:
|
|
The event ID
|
|
"""
|
|
try:
|
|
event_dict = event.to_dict()
|
|
event_dict['_seq'] = next(_tracking_event_seq)
|
|
tracking_events_db.insert(event_dict)
|
|
logger.info(f"Tracking event created: {event.action} {event.entity_type} {event.entity_id} for child {event.child_id}")
|
|
return event.id
|
|
except Exception as e:
|
|
logger.error(f"Failed to insert tracking event: {e}")
|
|
raise
|
|
|
|
|
|
def get_tracking_events_by_child(
|
|
child_id: str,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
entity_type: Optional[EntityType] = None,
|
|
action: Optional[ActionType] = None
|
|
) -> tuple[List[TrackingEvent], int]:
|
|
"""
|
|
Query tracking events for a specific child with optional filters.
|
|
|
|
Args:
|
|
child_id: Child ID to filter by
|
|
limit: Maximum number of results (default 50, max 500)
|
|
offset: Number of results to skip
|
|
entity_type: Optional filter by entity type
|
|
action: Optional filter by action type
|
|
|
|
Returns:
|
|
Tuple of (list of TrackingEvent instances, total count)
|
|
"""
|
|
limit = min(limit, 500)
|
|
|
|
TrackingQuery = Query()
|
|
query_condition = TrackingQuery.child_id == child_id
|
|
|
|
if entity_type:
|
|
query_condition &= TrackingQuery.entity_type == entity_type
|
|
if action:
|
|
query_condition &= TrackingQuery.action == action
|
|
|
|
all_results = tracking_events_db.search(query_condition)
|
|
total = len(all_results)
|
|
|
|
# Sort by occurred_at desc, then created_at desc, then _seq desc for
|
|
# deterministic ordering when timestamps collide (common in fast tests).
|
|
all_results.sort(
|
|
key=lambda x: (x.get('occurred_at', ''), x.get('created_at', 0), x.get('_seq', 0)),
|
|
reverse=True,
|
|
)
|
|
|
|
paginated = all_results[offset:offset + limit]
|
|
events = [TrackingEvent.from_dict(r) for r in paginated]
|
|
|
|
return events, total
|
|
|
|
|
|
def get_tracking_events_by_user(
|
|
user_id: str,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
entity_type: Optional[EntityType] = None
|
|
) -> tuple[List[TrackingEvent], int]:
|
|
"""
|
|
Query tracking events for a specific user.
|
|
|
|
Args:
|
|
user_id: User ID to filter by
|
|
limit: Maximum number of results
|
|
offset: Number of results to skip
|
|
entity_type: Optional filter by entity type
|
|
|
|
Returns:
|
|
Tuple of (list of TrackingEvent instances, total count)
|
|
"""
|
|
limit = min(limit, 500)
|
|
|
|
TrackingQuery = Query()
|
|
query_condition = TrackingQuery.user_id == user_id
|
|
|
|
if entity_type:
|
|
query_condition &= TrackingQuery.entity_type == entity_type
|
|
|
|
all_results = tracking_events_db.search(query_condition)
|
|
total = len(all_results)
|
|
|
|
all_results.sort(
|
|
key=lambda x: (x.get('occurred_at', ''), x.get('created_at', 0), x.get('_seq', 0)),
|
|
reverse=True,
|
|
)
|
|
|
|
paginated = all_results[offset:offset + limit]
|
|
events = [TrackingEvent.from_dict(r) for r in paginated]
|
|
|
|
return events, total
|
|
|
|
|
|
def anonymize_tracking_events_for_user(user_id: str) -> int:
|
|
"""
|
|
Anonymize tracking events by setting user_id to None.
|
|
Called when a user is deleted.
|
|
|
|
Args:
|
|
user_id: User ID to anonymize
|
|
|
|
Returns:
|
|
Number of records anonymized
|
|
"""
|
|
TrackingQuery = Query()
|
|
result = tracking_events_db.update({'user_id': None}, TrackingQuery.user_id == user_id)
|
|
count = len(result) if result else 0
|
|
logger.info(f"Anonymized {count} tracking events for user {user_id}")
|
|
return count
|