186 lines
6.5 KiB
Python
186 lines
6.5 KiB
Python
"""Seed test data into the database for demo purposes."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Add parent directory to path to import baby_monitor modules
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
# Import after path is set up
|
|
from baby_monitor.repositories.dependencies.get_database import ( # noqa: E402
|
|
get_database,
|
|
init_db,
|
|
)
|
|
from baby_monitor.repositories.user.sqlite_user import ( # noqa: E402
|
|
DatabaseUserRepository,
|
|
)
|
|
from baby_monitor.repositories.child.sqlite_child import ( # noqa: E402
|
|
DatabaseChildRepository,
|
|
)
|
|
from baby_monitor.repositories.feeding.sqlite_feeding import ( # noqa: E402
|
|
DatabaseFeedingRepository,
|
|
)
|
|
from baby_monitor.repositories.diaper_change.sqlite_diaper_change import ( # noqa: E402, E501
|
|
DatabaseDiaperChangeRepository,
|
|
)
|
|
from baby_monitor.repositories.sleep.sqlite_sleep import ( # noqa: E402
|
|
DatabaseSleepRepository,
|
|
)
|
|
from baby_monitor.routers.auth import hash_password # noqa: E402
|
|
from showcase_data_definitions import generate_showcase_data # noqa: E402
|
|
|
|
|
|
def seed_database(reference_time: datetime | None = None) -> None:
|
|
"""
|
|
Seed the database with test data.
|
|
|
|
This function is idempotent - it will check if test users exist
|
|
and either create them or update timestamps of existing data.
|
|
|
|
Args:
|
|
reference_time: The "now" time to use for generating timestamps.
|
|
Defaults to current time.
|
|
"""
|
|
# Initialize database
|
|
init_db()
|
|
db = next(get_database())
|
|
|
|
try:
|
|
# Initialize repositories
|
|
user_repo = DatabaseUserRepository(db)
|
|
child_repo = DatabaseChildRepository(db)
|
|
feeding_repo = DatabaseFeedingRepository(db)
|
|
diaper_repo = DatabaseDiaperChangeRepository(db)
|
|
sleep_repo = DatabaseSleepRepository(db)
|
|
|
|
# Generate showcase data
|
|
test_data = generate_showcase_data(reference_time)
|
|
|
|
print("Starting database seeding...")
|
|
print(f"Reference time: {test_data['reference_time']}")
|
|
|
|
# Create or get showcase users
|
|
user_ids = []
|
|
for user_data in test_data["users"]:
|
|
# Check if user exists
|
|
existing_user = user_repo.get_by_username(user_data["username"])
|
|
|
|
if existing_user:
|
|
print(f"✓ User '{user_data['username']}' already exists")
|
|
user_ids.append(existing_user["id"])
|
|
else:
|
|
# Create new user
|
|
hashed_pw = hash_password(user_data["password"])
|
|
user_id = user_repo.create(
|
|
username=user_data["username"],
|
|
hashed_password=hashed_pw,
|
|
)
|
|
user_ids.append(user_id["id"])
|
|
print(f"✓ Created user '{user_data['username']}'")
|
|
|
|
# Create or get shared child
|
|
child_data = test_data["child"]
|
|
|
|
# Check if child exists for first user
|
|
existing_children = child_repo.get_by_user_id(user_ids[0])
|
|
child_id: int | None = None
|
|
|
|
# Look for existing child with matching name
|
|
for child in existing_children:
|
|
if child["name"] == child_data["name"]:
|
|
child_id = child["id"]
|
|
print(f"✓ Child '{child_data['name']}' already exists")
|
|
break
|
|
|
|
if child_id is None:
|
|
# Create new child
|
|
created_child = child_repo.create(
|
|
name=child_data["name"],
|
|
birth_time=child_data["birth_time"],
|
|
birth_weight=child_data["birth_weight"],
|
|
user_id=user_ids[0],
|
|
)
|
|
child_id = created_child["id"]
|
|
print(f"✓ Created child '{child_data['name']}'")
|
|
|
|
# Ensure child_id is set
|
|
assert child_id is not None, "Child ID must be set at this point"
|
|
|
|
# Share child with second user if not already shared
|
|
parent_ids = child_repo.get_parent_ids(child_id)
|
|
if user_ids[1] not in parent_ids:
|
|
child_repo.add_parent(child_id, user_ids[1])
|
|
msg = f"✓ Shared child with '{test_data['users'][1]['username']}'"
|
|
print(msg)
|
|
|
|
# Clear existing logs to avoid duplicates
|
|
print("Clearing existing logs...")
|
|
|
|
# Get all logs for this child
|
|
existing_feedings = feeding_repo.get_by_child_id(child_id)
|
|
for feeding in existing_feedings:
|
|
feeding_repo.delete(feeding["id"])
|
|
|
|
existing_diapers = diaper_repo.get_by_child_id(child_id)
|
|
for diaper in existing_diapers:
|
|
diaper_repo.delete(diaper["id"])
|
|
|
|
existing_sleep = sleep_repo.get_by_child_id(child_id)
|
|
for sleep in existing_sleep:
|
|
sleep_repo.delete(sleep["id"])
|
|
|
|
print("✓ Cleared existing logs")
|
|
|
|
# Add feeding logs
|
|
print(f"Adding {len(test_data['feedings'])} feeding logs...")
|
|
for feeding in test_data["feedings"]:
|
|
feeding_repo.create(
|
|
child_id=child_id,
|
|
start_time=feeding["start_time"],
|
|
end_time=feeding["end_time"],
|
|
feeding_type=feeding["feeding_type"],
|
|
)
|
|
print(f"✓ Added {len(test_data['feedings'])} feedings")
|
|
|
|
# Add diaper changes
|
|
print(f"Adding {len(test_data['diapers'])} diaper changes...")
|
|
for diaper in test_data["diapers"]:
|
|
diaper_repo.create(
|
|
child_id=child_id,
|
|
change_time=diaper["time"],
|
|
poop_color=diaper.get("poop_color"),
|
|
poop_amount=diaper.get("poop_amount"),
|
|
pee_color=diaper.get("pee_color"),
|
|
pee_amount=diaper.get("pee_amount"),
|
|
)
|
|
print(f"✓ Added {len(test_data['diapers'])} diaper changes")
|
|
|
|
# Add sleep sessions
|
|
print(f"Adding {len(test_data['sleep'])} sleep sessions...")
|
|
for sleep in test_data["sleep"]:
|
|
sleep_repo.create(
|
|
child_id=child_id,
|
|
start_time=sleep["start_time"],
|
|
end_time=sleep["end_time"],
|
|
)
|
|
print(f"✓ Added {len(test_data['sleep'])} sleep sessions")
|
|
|
|
db.commit()
|
|
print("\n✅ Database seeding completed successfully!")
|
|
print("\nTest user credentials:")
|
|
print(f" Username: {test_data['users'][0]['username']}")
|
|
print(f" Password: {test_data['users'][0]['password']}")
|
|
print(" (Both users share the same password)")
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
print(f"\n❌ Error seeding database: {e}")
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
seed_database()
|