80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, select, func
|
|
from database import get_session
|
|
from models import User, Review, Attendance, Group, GroupMember, Show
|
|
from schemas import UserRead, ReviewRead, ShowRead, GroupRead
|
|
from auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
# --- User Stats ---
|
|
|
|
@router.get("/{user_id}/stats")
|
|
def get_user_stats(user_id: int, session: Session = Depends(get_session)):
|
|
# Check if user exists
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
attendance_count = session.exec(select(func.count(Attendance.id)).where(Attendance.user_id == user_id)).one()
|
|
review_count = session.exec(select(func.count(Review.id)).where(Review.user_id == user_id)).one()
|
|
group_count = session.exec(select(func.count(GroupMember.id)).where(GroupMember.user_id == user_id)).one()
|
|
|
|
return {
|
|
"attendance_count": attendance_count,
|
|
"review_count": review_count,
|
|
"group_count": group_count
|
|
}
|
|
|
|
# --- User Data Lists ---
|
|
|
|
@router.get("/{user_id}/attendance", response_model=List[ShowRead])
|
|
def get_user_attendance(
|
|
user_id: int,
|
|
offset: int = 0,
|
|
limit: int = Query(default=50, le=100),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
# Return shows the user attended
|
|
shows = session.exec(
|
|
select(Show)
|
|
.join(Attendance, Show.id == Attendance.show_id)
|
|
.where(Attendance.user_id == user_id)
|
|
.order_by(Show.date.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
).all()
|
|
return shows
|
|
|
|
@router.get("/{user_id}/reviews", response_model=List[ReviewRead])
|
|
def get_user_reviews(
|
|
user_id: int,
|
|
offset: int = 0,
|
|
limit: int = Query(default=50, le=100),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
reviews = session.exec(
|
|
select(Review)
|
|
.where(Review.user_id == user_id)
|
|
.order_by(Review.created_at.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
).all()
|
|
return reviews
|
|
|
|
@router.get("/{user_id}/groups", response_model=List[GroupRead])
|
|
def get_user_groups(
|
|
user_id: int,
|
|
offset: int = 0,
|
|
limit: int = Query(default=50, le=100),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
groups = session.exec(
|
|
select(Group)
|
|
.join(GroupMember, Group.id == GroupMember.group_id)
|
|
.where(GroupMember.user_id == user_id)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
).all()
|
|
return groups
|