added sleep logging workflow and overview page
Build and Push Docker Image / build-and-push (pull_request) Successful in 58s
Python Code Quality / python-code-quality (pull_request) Failing after 10s
Python Test / python-test (pull_request) Successful in 17s

This commit is contained in:
Brian Bjarke Jensen
2025-11-10 21:00:03 +01:00
parent edeb01c889
commit cd3db4fb86
12 changed files with 1554 additions and 0 deletions
+155
View File
@@ -0,0 +1,155 @@
"""Sleep log management router."""
from typing import Annotated
from fastapi import APIRouter, HTTPException, Depends
from baby_monitor.models.sleep import (
CreateSleepRequest,
UpdateSleepRequest,
SleepResponse,
)
from baby_monitor.routers.auth import verify_token
from baby_monitor.repositories.sleep.sqlite_sleep import SQLiteSleepRepository
from baby_monitor.repositories.dependencies.get_sleep_repository import (
get_sleep_repository,
)
from baby_monitor.repositories.child.sqlite_child import SQLiteChildRepository
from baby_monitor.repositories.dependencies.get_child_repository import (
get_child_repository,
)
router = APIRouter(prefix="/api/sleep", tags=["sleep"])
@router.post("", response_model=SleepResponse, status_code=201)
def create_sleep(
request: CreateSleepRequest,
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
child_repo: Annotated[SQLiteChildRepository, Depends(get_child_repository)],
) -> SleepResponse:
"""Create a new sleep log entry."""
# Verify the child belongs to the authenticated user
child = child_repo.get_by_id(request.child_id)
if not child:
raise HTTPException(status_code=404, detail="Child not found")
if child["user_id"] != user_id:
raise HTTPException(
status_code=403, detail="Access denied to this child"
)
sleep = sleep_repo.create(
child_id=request.child_id,
start_time=request.start_time,
)
return SleepResponse(**sleep)
@router.get("/active", response_model=SleepResponse | None)
def get_active_sleep(
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
child_repo: Annotated[SQLiteChildRepository, Depends(get_child_repository)],
) -> SleepResponse | None:
"""Get the current active sleep (where end_time is null) for the user."""
sleeps = sleep_repo.get_by_user_id(user_id)
# Find the first sleep with no end_time
for sleep in sleeps:
if sleep["end_time"] is None:
# Get child info to include in response
child = child_repo.get_by_id(sleep["child_id"])
response_data = sleep.copy()
if child:
response_data["child_name"] = child["name"]
return SleepResponse(**response_data)
return None
@router.get("", response_model=list[SleepResponse])
def get_user_sleeps(
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
) -> list[SleepResponse]:
"""Get all sleep logs for the authenticated user's children."""
sleeps = sleep_repo.get_by_user_id(user_id)
return [SleepResponse(**sleep) for sleep in sleeps]
@router.get("/{sleep_id}", response_model=SleepResponse)
def get_sleep(
sleep_id: int,
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
child_repo: Annotated[SQLiteChildRepository, Depends(get_child_repository)],
) -> SleepResponse:
"""Get a specific sleep log by ID."""
sleep = sleep_repo.get_by_id(sleep_id)
if not sleep:
raise HTTPException(status_code=404, detail="Sleep log not found")
# Verify the sleep belongs to user's child
child = child_repo.get_by_id(sleep["child_id"])
if not child or child["user_id"] != user_id:
raise HTTPException(status_code=403, detail="Access denied")
return SleepResponse(**sleep)
@router.put("/{sleep_id}", response_model=SleepResponse)
def update_sleep(
sleep_id: int,
request: UpdateSleepRequest,
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
child_repo: Annotated[SQLiteChildRepository, Depends(get_child_repository)],
) -> SleepResponse:
"""Update an existing sleep log."""
sleep = sleep_repo.get_by_id(sleep_id)
if not sleep:
raise HTTPException(status_code=404, detail="Sleep log not found")
# Verify the sleep belongs to user's child
child = child_repo.get_by_id(sleep["child_id"])
if not child or child["user_id"] != user_id:
raise HTTPException(status_code=403, detail="Access denied")
updated = sleep_repo.update(
sleep_id=sleep_id,
end_time=request.end_time,
)
if not updated:
raise HTTPException(
status_code=500, detail="Failed to update sleep log"
)
return SleepResponse(**updated)
@router.delete("/{sleep_id}", status_code=204)
def delete_sleep(
sleep_id: int,
user_id: Annotated[int, Depends(verify_token)],
sleep_repo: Annotated[SQLiteSleepRepository, Depends(get_sleep_repository)],
child_repo: Annotated[SQLiteChildRepository, Depends(get_child_repository)],
) -> None:
"""Delete a sleep log."""
sleep = sleep_repo.get_by_id(sleep_id)
if not sleep:
raise HTTPException(status_code=404, detail="Sleep log not found")
# Verify the sleep belongs to user's child
child = child_repo.get_by_id(sleep["child_id"])
if not child or child["user_id"] != user_id:
raise HTTPException(status_code=403, detail="Access denied")
success = sleep_repo.delete(sleep_id)
if not success:
raise HTTPException(status_code=500, detail="Failed to delete sleep log")