Ryan Kegel
3dee8b80a2
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 24s
- Added tracking events for tasks, penalties, and rewards with timestamps. - Created new TinyDB table for tracking records to maintain audit history. - Developed backend API for querying tracking events with filters and pagination. - Implemented logging for tracking events with per-user rotating log files. - Added unit tests for tracking event creation, querying, and anonymization. - Deferred frontend changes for future implementation. - Established acceptance criteria and documentation for the tracking feature. feat: Introduce account deletion scheduler - Implemented a scheduler to delete accounts marked for deletion after a configurable threshold. - Added new fields to the User model to manage deletion status and attempts. - Created admin API endpoints for managing deletion thresholds and viewing the deletion queue. - Integrated error handling and logging for the deletion process. - Developed unit tests for the deletion scheduler and related API endpoints. - Documented the deletion process and acceptance criteria.
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Literal, Optional
|
|
from models.base import BaseModel
|
|
|
|
|
|
EntityType = Literal['task', 'reward', 'penalty']
|
|
ActionType = Literal['activated', 'requested', 'redeemed', 'cancelled']
|
|
|
|
|
|
@dataclass
|
|
class TrackingEvent(BaseModel):
|
|
user_id: Optional[str]
|
|
child_id: str
|
|
entity_type: EntityType
|
|
entity_id: str
|
|
action: ActionType
|
|
points_before: int
|
|
points_after: int
|
|
delta: int
|
|
occurred_at: str # UTC ISO 8601 timestamp
|
|
metadata: Optional[dict] = None
|
|
|
|
def __post_init__(self):
|
|
"""Validate invariants after initialization."""
|
|
if self.delta != self.points_after - self.points_before:
|
|
raise ValueError(
|
|
f"Delta invariant violated: {self.delta} != {self.points_after} - {self.points_before}"
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict):
|
|
return cls(
|
|
user_id=d.get('user_id'),
|
|
child_id=d.get('child_id'),
|
|
entity_type=d.get('entity_type'),
|
|
entity_id=d.get('entity_id'),
|
|
action=d.get('action'),
|
|
points_before=d.get('points_before'),
|
|
points_after=d.get('points_after'),
|
|
delta=d.get('delta'),
|
|
occurred_at=d.get('occurred_at'),
|
|
metadata=d.get('metadata'),
|
|
id=d.get('id'),
|
|
created_at=d.get('created_at'),
|
|
updated_at=d.get('updated_at')
|
|
)
|
|
|
|
def to_dict(self):
|
|
base = super().to_dict()
|
|
base.update({
|
|
'user_id': self.user_id,
|
|
'child_id': self.child_id,
|
|
'entity_type': self.entity_type,
|
|
'entity_id': self.entity_id,
|
|
'action': self.action,
|
|
'points_before': self.points_before,
|
|
'points_after': self.points_after,
|
|
'delta': self.delta,
|
|
'occurred_at': self.occurred_at,
|
|
'metadata': self.metadata
|
|
})
|
|
return base
|
|
|
|
@staticmethod
|
|
def create_event(
|
|
user_id: Optional[str],
|
|
child_id: str,
|
|
entity_type: EntityType,
|
|
entity_id: str,
|
|
action: ActionType,
|
|
points_before: int,
|
|
points_after: int,
|
|
metadata: Optional[dict] = None
|
|
) -> 'TrackingEvent':
|
|
"""Factory method to create a tracking event with server timestamp."""
|
|
delta = points_after - points_before
|
|
occurred_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
return TrackingEvent(
|
|
user_id=user_id,
|
|
child_id=child_id,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
action=action,
|
|
points_before=points_before,
|
|
points_after=points_after,
|
|
delta=delta,
|
|
occurred_at=occurred_at,
|
|
metadata=metadata
|
|
)
|