92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, select, func
|
|
|
|
from app.database import get_session
|
|
from app.models.user import User
|
|
from app.models.zone import Zone
|
|
from app.models.activity import Activity
|
|
from app.models.score import Score
|
|
from app.schemas.user import UserRead, UserUpdate, UserStats
|
|
from app.auth.dependencies import get_current_user
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
@router.get("/me", response_model=UserRead)
|
|
def get_me(current_user: User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
@router.patch("/me", response_model=UserRead)
|
|
def update_me(
|
|
body: UserUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
if body.username is not None:
|
|
existing = session.exec(
|
|
select(User).where(
|
|
User.username == body.username, User.id != current_user.id
|
|
)
|
|
).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Username already taken")
|
|
current_user.username = body.username
|
|
if body.avatar_url is not None:
|
|
current_user.avatar_url = body.avatar_url
|
|
if body.fcm_token is not None:
|
|
current_user.fcm_token = body.fcm_token
|
|
|
|
session.add(current_user)
|
|
session.commit()
|
|
session.refresh(current_user)
|
|
return current_user
|
|
|
|
|
|
@router.get("/{user_id}/stats", response_model=UserStats)
|
|
def get_user_stats(
|
|
user_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Total area
|
|
total_area = session.exec(
|
|
select(func.coalesce(func.sum(Zone.area_m2), 0.0)).where(
|
|
Zone.owner_id == user_id
|
|
)
|
|
).one()
|
|
|
|
# Zone count
|
|
zone_count = session.exec(
|
|
select(func.count()).select_from(Zone).where(Zone.owner_id == user_id)
|
|
).one()
|
|
|
|
# Activity count
|
|
activity_count = session.exec(
|
|
select(func.count())
|
|
.select_from(Activity)
|
|
.where(Activity.user_id == user_id, Activity.status == "completed")
|
|
).one()
|
|
|
|
# Total points (latest score)
|
|
latest_score = session.exec(
|
|
select(Score).where(Score.user_id == user_id).order_by(Score.date.desc()) # type: ignore[union-attr]
|
|
).first()
|
|
total_pts = latest_score.total_pts if latest_score else 0
|
|
|
|
return UserStats(
|
|
id=user.id,
|
|
username=user.username,
|
|
avatar_url=user.avatar_url,
|
|
total_area_m2=float(total_area),
|
|
total_points=total_pts,
|
|
zone_count=zone_count,
|
|
activity_count=activity_count,
|
|
)
|