3edbcdeb64
- Add slug fields to Song, Venue, Show, Tour, Performance models - Update routers to support lookup by slug or ID - Create slugify.py utility for generating URL-safe slugs - Add migration script to generate slugs for existing data - Performance slugs use songslug-YYYY-MM-DD format
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
from typing import List
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, select
|
|
from database import get_session
|
|
from models import Song, User, Tag, EntityTag
|
|
from schemas import SongCreate, SongRead, SongReadWithStats, SongUpdate, TagRead
|
|
from auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/songs", tags=["songs"])
|
|
|
|
@router.post("/", response_model=SongRead)
|
|
def create_song(song: SongCreate, session: Session = Depends(get_session), current_user = Depends(get_current_user)):
|
|
db_song = Song.model_validate(song)
|
|
session.add(db_song)
|
|
session.commit()
|
|
session.refresh(db_song)
|
|
return db_song
|
|
|
|
@router.get("/", response_model=List[SongRead])
|
|
def read_songs(offset: int = 0, limit: int = Query(default=100, le=100), session: Session = Depends(get_session)):
|
|
songs = session.exec(select(Song).offset(offset).limit(limit)).all()
|
|
return songs
|
|
|
|
from services.stats import get_song_stats
|
|
|
|
@router.get("/{song_id_or_slug}", response_model=SongReadWithStats)
|
|
def read_song(song_id_or_slug: str, session: Session = Depends(get_session)):
|
|
# Try to parse as int (ID), otherwise treat as slug
|
|
song = None
|
|
if song_id_or_slug.isdigit():
|
|
song = session.get(Song, int(song_id_or_slug))
|
|
|
|
if not song:
|
|
# Try slug lookup
|
|
song = session.exec(select(Song).where(Song.slug == song_id_or_slug)).first()
|
|
|
|
if not song:
|
|
raise HTTPException(status_code=404, detail="Song not found")
|
|
|
|
song_id = song.id # Use actual ID for lookups
|
|
stats = get_song_stats(session, song_id)
|
|
|
|
tags = session.exec(
|
|
select(Tag)
|
|
.join(EntityTag, Tag.id == EntityTag.tag_id)
|
|
.where(EntityTag.entity_type == "song")
|
|
.where(EntityTag.entity_id == song_id)
|
|
).all()
|
|
|
|
# Fetch performances
|
|
# We join Show to ensure we can order by date
|
|
from models import Show, Performance, Rating
|
|
from sqlmodel import func
|
|
# We need PerformanceReadWithShow from schemas
|
|
from schemas import PerformanceReadWithShow
|
|
|
|
perfs = session.exec(
|
|
select(Performance)
|
|
.join(Show)
|
|
.where(Performance.song_id == song_id)
|
|
.order_by(Show.date.desc())
|
|
).all()
|
|
|
|
# Calculate ratings
|
|
perf_ids = [p.id for p in perfs]
|
|
rating_stats = {}
|
|
if perf_ids:
|
|
results = session.exec(
|
|
select(Rating.performance_id, func.avg(Rating.score), func.count(Rating.id))
|
|
.where(Rating.performance_id.in_(perf_ids))
|
|
.group_by(Rating.performance_id)
|
|
).all()
|
|
for r in results:
|
|
rating_stats[r[0]] = {"avg": float(r[1]) if r[1] else 0.0, "count": r[2]}
|
|
|
|
perf_dtos = []
|
|
for p in perfs:
|
|
# Lazy load show/venue (could be optimized)
|
|
venue_name = "Unknown"
|
|
venue_city = ""
|
|
venue_state = ""
|
|
show_date = datetime.now()
|
|
|
|
if p.show:
|
|
show_date = p.show.date
|
|
if p.show.venue:
|
|
venue_name = p.show.venue.name
|
|
venue_city = p.show.venue.city
|
|
venue_state = p.show.venue.state
|
|
|
|
r_stats = rating_stats.get(p.id, {"avg": 0.0, "count": 0})
|
|
|
|
perf_dtos.append(PerformanceReadWithShow(
|
|
**p.model_dump(),
|
|
show_date=show_date,
|
|
venue_name=venue_name,
|
|
venue_city=venue_city,
|
|
venue_state=venue_state,
|
|
avg_rating=r_stats["avg"],
|
|
total_reviews=r_stats["count"]
|
|
))
|
|
|
|
# Merge song data with stats
|
|
song_with_stats = SongReadWithStats(
|
|
**song.model_dump(),
|
|
**stats
|
|
)
|
|
song_with_stats.tags = tags
|
|
song_with_stats.performances = perf_dtos
|
|
return song_with_stats
|
|
|
|
@router.patch("/{song_id}", response_model=SongRead)
|
|
def update_song(song_id: int, song: SongUpdate, session: Session = Depends(get_session), current_user = Depends(get_current_user)):
|
|
db_song = session.get(Song, song_id)
|
|
if not db_song:
|
|
raise HTTPException(status_code=404, detail="Song not found")
|
|
song_data = song.model_dump(exclude_unset=True)
|
|
db_song.sqlmodel_update(song_data)
|
|
session.add(db_song)
|
|
session.commit()
|
|
session.refresh(db_song)
|
|
return db_song
|