256 lines
7.6 KiB
Python
256 lines
7.6 KiB
Python
"""Admin router for administrative functions."""
|
|
|
|
import secrets
|
|
from datetime import datetime, timedelta, UTC
|
|
from typing import Annotated
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from baby_monitor.routers.auth import verify_admin
|
|
from baby_monitor.repositories.dependencies.get_invitation_repository import (
|
|
get_invitation_repository,
|
|
)
|
|
from baby_monitor.repositories.dependencies.get_user_repository import (
|
|
get_user_repository,
|
|
)
|
|
from baby_monitor.repositories.dependencies.get_child_repository import (
|
|
get_child_repository,
|
|
)
|
|
from baby_monitor.repositories.dependencies.get_feeding_repository import (
|
|
get_feeding_repository,
|
|
)
|
|
from baby_monitor.repositories.dependencies.get_diaper_change_repository import (
|
|
get_diaper_change_repository,
|
|
)
|
|
from baby_monitor.repositories.dependencies.get_sleep_repository import (
|
|
get_sleep_repository,
|
|
)
|
|
from baby_monitor.repositories.interfaces import (
|
|
InvitationRepositoryInterface,
|
|
UserRepositoryInterface,
|
|
ChildRepositoryInterface,
|
|
FeedingRepositoryInterface,
|
|
DiaperChangeRepositoryInterface,
|
|
SleepRepositoryInterface,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
class InvitationResponse(BaseModel):
|
|
"""Response model for invitation link generation."""
|
|
|
|
token: str
|
|
expires_at: str
|
|
message: str
|
|
|
|
|
|
@router.post("/generate-invitation", response_model=InvitationResponse)
|
|
async def generate_invitation_link(
|
|
user_id: Annotated[int, Depends(verify_admin)],
|
|
invitation_repository: Annotated[
|
|
InvitationRepositoryInterface,
|
|
Depends(get_invitation_repository),
|
|
],
|
|
) -> InvitationResponse:
|
|
"""Generate a new invitation link for user registration.
|
|
|
|
The invitation token is valid for 24 hours and can be used once.
|
|
Requires admin role.
|
|
|
|
Args:
|
|
user_id: Admin user ID (from verify_admin)
|
|
invitation_repository: Invitation storage backend
|
|
|
|
Returns:
|
|
InvitationResponse with token and expiration details
|
|
"""
|
|
|
|
# Generate secure random token
|
|
invitation_token = secrets.token_urlsafe(32)
|
|
|
|
# Calculate expiration (24 hours from now)
|
|
expires_at = datetime.now(UTC) + timedelta(hours=24)
|
|
|
|
# Store invitation token in database
|
|
invitation_repository.create_invitation(
|
|
token=invitation_token,
|
|
created_by_user_id=user_id,
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
return InvitationResponse(
|
|
token=invitation_token,
|
|
expires_at=expires_at.isoformat(),
|
|
message="Invitation link generated successfully. Valid for 24 hours.",
|
|
)
|
|
|
|
|
|
class UserResponse(BaseModel):
|
|
"""Response model for user data."""
|
|
|
|
id: int
|
|
username: str
|
|
is_admin: bool
|
|
created_at: datetime
|
|
|
|
|
|
class DeleteUserResponse(BaseModel):
|
|
"""Response model for user deletion."""
|
|
|
|
message: str
|
|
deleted_children: list[int]
|
|
deleted_feedings: int
|
|
deleted_diapers: int
|
|
deleted_sleeps: int
|
|
|
|
|
|
@router.get("/users", response_model=list[UserResponse])
|
|
async def get_all_users(
|
|
admin_id: Annotated[int, Depends(verify_admin)],
|
|
user_repository: Annotated[
|
|
UserRepositoryInterface,
|
|
Depends(get_user_repository),
|
|
],
|
|
) -> list[UserResponse]:
|
|
"""Get all users in the system.
|
|
|
|
Requires admin role.
|
|
|
|
Args:
|
|
admin_id: Admin user ID (from verify_admin)
|
|
user_repository: User storage backend
|
|
|
|
Returns:
|
|
List of all users with their basic information
|
|
"""
|
|
users = user_repository.get_all()
|
|
return [
|
|
UserResponse(
|
|
id=user["id"],
|
|
username=user["username"],
|
|
is_admin=user["is_admin"],
|
|
created_at=user["created_at"],
|
|
)
|
|
for user in users
|
|
]
|
|
|
|
|
|
@router.delete("/users/{user_id}", response_model=DeleteUserResponse)
|
|
async def delete_user(
|
|
user_id: int,
|
|
admin_id: Annotated[int, Depends(verify_admin)],
|
|
user_repository: Annotated[
|
|
UserRepositoryInterface,
|
|
Depends(get_user_repository),
|
|
],
|
|
child_repository: Annotated[
|
|
ChildRepositoryInterface,
|
|
Depends(get_child_repository),
|
|
],
|
|
feeding_repository: Annotated[
|
|
FeedingRepositoryInterface,
|
|
Depends(get_feeding_repository),
|
|
],
|
|
diaper_repository: Annotated[
|
|
DiaperChangeRepositoryInterface,
|
|
Depends(get_diaper_change_repository),
|
|
],
|
|
sleep_repository: Annotated[
|
|
SleepRepositoryInterface,
|
|
Depends(get_sleep_repository),
|
|
],
|
|
) -> DeleteUserResponse:
|
|
"""Delete a user and cascade delete orphaned children and their logs.
|
|
|
|
When a user is deleted:
|
|
1. Find all children belonging to this user
|
|
2. For each child, check if other users also have this child
|
|
3. Delete only children that have no other parents (orphaned)
|
|
4. For orphaned children, delete all associated logs (feedings, diapers, sleep)
|
|
5. Finally delete the user
|
|
|
|
Requires admin role.
|
|
|
|
Args:
|
|
user_id: ID of user to delete
|
|
admin_id: Admin user ID (from verify_admin)
|
|
user_repository: User storage backend
|
|
child_repository: Child storage backend
|
|
feeding_repository: Feeding storage backend
|
|
diaper_repository: Diaper change storage backend
|
|
sleep_repository: Sleep storage backend
|
|
|
|
Returns:
|
|
DeleteUserResponse with deletion summary
|
|
|
|
Raises:
|
|
HTTPException: If user not found or trying to delete yourself
|
|
"""
|
|
# Prevent admin from deleting themselves
|
|
if user_id == admin_id:
|
|
raise HTTPException(status_code=400, detail="Cannot delete your own account")
|
|
|
|
# Check if user exists
|
|
user = user_repository.get_by_id(user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Get all children belonging to this user
|
|
children = child_repository.get_by_user_id(user_id)
|
|
|
|
deleted_children_ids = []
|
|
total_feedings_deleted = 0
|
|
total_diapers_deleted = 0
|
|
total_sleeps_deleted = 0
|
|
|
|
# For each child, check if they have other parents
|
|
for child in children:
|
|
child_id = child["id"]
|
|
|
|
# Remove this user's relationship with the child
|
|
child_repository.remove_parent(child_id, user_id)
|
|
|
|
# Check if child has any remaining parents
|
|
remaining_parents = child_repository.get_parent_ids(child_id)
|
|
|
|
# Only delete child and logs if they have no other parents
|
|
if len(remaining_parents) == 0:
|
|
# Delete all logs for this orphaned child
|
|
feedings = feeding_repository.get_by_child_id(child_id)
|
|
for feeding in feedings:
|
|
feeding_repository.delete(feeding["id"])
|
|
total_feedings_deleted += 1
|
|
|
|
diapers = diaper_repository.get_by_child_id(child_id)
|
|
for diaper in diapers:
|
|
diaper_repository.delete(diaper["id"])
|
|
total_diapers_deleted += 1
|
|
|
|
sleeps = sleep_repository.get_by_child_id(child_id)
|
|
for sleep in sleeps:
|
|
sleep_repository.delete(sleep["id"])
|
|
total_sleeps_deleted += 1
|
|
|
|
# Delete the orphaned child
|
|
child_repository.delete(child_id)
|
|
deleted_children_ids.append(child_id)
|
|
|
|
# Finally, delete the user
|
|
user_repository.delete(user_id)
|
|
|
|
message = f"User '{user['username']}' deleted successfully"
|
|
if deleted_children_ids:
|
|
message += (
|
|
f" with {len(deleted_children_ids)} orphaned "
|
|
f"child{'ren' if len(deleted_children_ids) != 1 else ''}"
|
|
)
|
|
|
|
return DeleteUserResponse(
|
|
message=message,
|
|
deleted_children=deleted_children_ids,
|
|
deleted_feedings=total_feedings_deleted,
|
|
deleted_diapers=total_diapers_deleted,
|
|
deleted_sleeps=total_sleeps_deleted,
|
|
)
|