Ryan Kegel
401c21ad82
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 25s
- Implemented PendingRewardDialog for handling pending reward requests. - Created RewardConfirmDialog for confirming reward redemption. - Developed TaskConfirmDialog for task confirmation with child name display. test: add unit tests for ChildView and ParentView components - Added comprehensive tests for ChildView including task triggering and SSE event handling. - Implemented tests for ParentView focusing on override modal and SSE event management. test: add ScrollingList component tests - Created tests for ScrollingList to verify item fetching, loading states, and custom item classes. - Included tests for two-step click interactions and edit button display logic. - Moved toward hashed passwords.
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
# python
|
|
import os
|
|
from config.paths import get_database_dir
|
|
import threading
|
|
from tinydb import TinyDB
|
|
|
|
base_dir = get_database_dir()
|
|
os.makedirs(base_dir, exist_ok=True)
|
|
|
|
|
|
class LockedTable:
|
|
"""
|
|
Thread-safe wrapper around a TinyDB table. All callable attribute access
|
|
is wrapped to acquire a reentrant lock while calling the underlying method.
|
|
Non-callable attributes are returned directly.
|
|
"""
|
|
def __init__(self, table):
|
|
self._table = table
|
|
self._lock = threading.RLock()
|
|
|
|
def __getattr__(self, name):
|
|
# avoid proxying internal attrs
|
|
if name in ('_table', '_lock'):
|
|
return super().__getattribute__(name)
|
|
|
|
attr = getattr(self._table, name)
|
|
|
|
if callable(attr):
|
|
def locked_call(*args, **kwargs):
|
|
with self._lock:
|
|
return attr(*args, **kwargs)
|
|
return locked_call
|
|
return attr
|
|
|
|
# convenience explicit methods (ensure these are class methods, not top-level)
|
|
def insert(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.insert(*args, **kwargs)
|
|
|
|
def insert_multiple(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.insert_multiple(*args, **kwargs)
|
|
|
|
def search(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.search(*args, **kwargs)
|
|
|
|
def get(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.get(*args, **kwargs)
|
|
|
|
def all(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.all(*args, **kwargs)
|
|
|
|
def remove(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.remove(*args, **kwargs)
|
|
|
|
def update(self, *args, **kwargs):
|
|
with self._lock:
|
|
return self._table.update(*args, **kwargs)
|
|
|
|
def truncate(self):
|
|
with self._lock:
|
|
return self._table.truncate()
|
|
|
|
# Setup DB files next to this module
|
|
|
|
child_path = os.path.join(base_dir, 'children.json')
|
|
task_path = os.path.join(base_dir, 'tasks.json')
|
|
reward_path = os.path.join(base_dir, 'rewards.json')
|
|
image_path = os.path.join(base_dir, 'images.json')
|
|
pending_reward_path = os.path.join(base_dir, 'pending_rewards.json')
|
|
users_path = os.path.join(base_dir, 'users.json')
|
|
tracking_events_path = os.path.join(base_dir, 'tracking_events.json')
|
|
child_overrides_path = os.path.join(base_dir, 'child_overrides.json')
|
|
|
|
# Use separate TinyDB instances/files for each collection
|
|
_child_db = TinyDB(child_path, indent=2)
|
|
_task_db = TinyDB(task_path, indent=2)
|
|
_reward_db = TinyDB(reward_path, indent=2)
|
|
_image_db = TinyDB(image_path, indent=2)
|
|
_pending_rewards_db = TinyDB(pending_reward_path, indent=2)
|
|
_users_db = TinyDB(users_path, indent=2)
|
|
_tracking_events_db = TinyDB(tracking_events_path, indent=2)
|
|
_child_overrides_db = TinyDB(child_overrides_path, indent=2)
|
|
|
|
# Expose table objects wrapped with locking
|
|
child_db = LockedTable(_child_db)
|
|
task_db = LockedTable(_task_db)
|
|
reward_db = LockedTable(_reward_db)
|
|
image_db = LockedTable(_image_db)
|
|
pending_reward_db = LockedTable(_pending_rewards_db)
|
|
users_db = LockedTable(_users_db)
|
|
tracking_events_db = LockedTable(_tracking_events_db)
|
|
child_overrides_db = LockedTable(_child_overrides_db)
|
|
|
|
if os.environ.get('DB_ENV', 'prod') == 'test':
|
|
child_db.truncate()
|
|
task_db.truncate()
|
|
reward_db.truncate()
|
|
image_db.truncate()
|
|
pending_reward_db.truncate()
|
|
users_db.truncate()
|
|
tracking_events_db.truncate()
|
|
child_overrides_db.truncate()
|
|
|