b4cddf41ea
- Fork elmeg-demo codebase for multi-band support - Add data importer infrastructure with base class - Create band-specific importers: - phish.py: Phish.net API v5 - grateful_dead.py: Grateful Stats API - setlistfm.py: Dead & Company, Billy Strings (Setlist.fm) - Add spec-kit configuration for Gemini - Update README with supported bands and architecture
138 lines
4.2 KiB
Python
138 lines
4.2 KiB
Python
"""
|
|
Musicians Router - API endpoints for managing musicians and band memberships.
|
|
"""
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, select
|
|
from pydantic import BaseModel
|
|
from database import get_session
|
|
from models import Musician, BandMembership, PerformanceGuest, Artist, Performance
|
|
from slugify import generate_slug as slugify
|
|
|
|
router = APIRouter(prefix="/musicians", tags=["musicians"])
|
|
|
|
# --- Schemas ---
|
|
class MusicianRead(BaseModel):
|
|
id: int
|
|
name: str
|
|
slug: str
|
|
bio: Optional[str] = None
|
|
image_url: Optional[str] = None
|
|
primary_instrument: Optional[str] = None
|
|
|
|
class MusicianCreate(BaseModel):
|
|
name: str
|
|
bio: Optional[str] = None
|
|
image_url: Optional[str] = None
|
|
primary_instrument: Optional[str] = None
|
|
|
|
class BandMembershipRead(BaseModel):
|
|
id: int
|
|
musician_id: int
|
|
artist_id: int
|
|
artist_name: Optional[str] = None
|
|
role: Optional[str] = None
|
|
start_date: Optional[str] = None
|
|
end_date: Optional[str] = None
|
|
|
|
class GuestAppearanceRead(BaseModel):
|
|
id: int
|
|
performance_id: int
|
|
song_title: Optional[str] = None
|
|
show_date: Optional[str] = None
|
|
instrument: Optional[str] = None
|
|
|
|
# --- Public Endpoints ---
|
|
|
|
@router.get("", response_model=List[MusicianRead])
|
|
def list_musicians(
|
|
search: Optional[str] = None,
|
|
limit: int = 50,
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""List all musicians"""
|
|
query = select(Musician)
|
|
if search:
|
|
query = query.where(Musician.name.icontains(search))
|
|
query = query.order_by(Musician.name).limit(limit)
|
|
return session.exec(query).all()
|
|
|
|
@router.get("/{slug}")
|
|
def get_musician(slug: str, session: Session = Depends(get_session)):
|
|
"""Get musician details with band memberships and guest appearances"""
|
|
musician = session.exec(select(Musician).where(Musician.slug == slug)).first()
|
|
if not musician:
|
|
raise HTTPException(status_code=404, detail="Musician not found")
|
|
|
|
# Get band memberships
|
|
memberships = session.exec(
|
|
select(BandMembership).where(BandMembership.musician_id == musician.id)
|
|
).all()
|
|
|
|
bands = []
|
|
for m in memberships:
|
|
artist = session.get(Artist, m.artist_id)
|
|
bands.append({
|
|
"id": m.id,
|
|
"artist_id": m.artist_id,
|
|
"artist_name": artist.name if artist else None,
|
|
"artist_slug": artist.slug if artist else None,
|
|
"role": m.role,
|
|
"start_date": str(m.start_date) if m.start_date else None,
|
|
"end_date": str(m.end_date) if m.end_date else None,
|
|
})
|
|
|
|
# Get guest appearances
|
|
appearances = session.exec(
|
|
select(PerformanceGuest).where(PerformanceGuest.musician_id == musician.id)
|
|
).all()
|
|
|
|
guests = []
|
|
for g in appearances:
|
|
perf = session.get(Performance, g.performance_id)
|
|
guests.append({
|
|
"id": g.id,
|
|
"performance_id": g.performance_id,
|
|
"performance_slug": perf.slug if perf else None,
|
|
"instrument": g.instrument,
|
|
})
|
|
|
|
return {
|
|
"musician": {
|
|
"id": musician.id,
|
|
"name": musician.name,
|
|
"slug": musician.slug,
|
|
"bio": musician.bio,
|
|
"image_url": musician.image_url,
|
|
"primary_instrument": musician.primary_instrument,
|
|
},
|
|
"bands": bands,
|
|
"guest_appearances": guests,
|
|
}
|
|
|
|
# --- Admin Endpoints (for now, no auth check - can be added later) ---
|
|
|
|
@router.post("", response_model=MusicianRead)
|
|
def create_musician(
|
|
musician_data: MusicianCreate,
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Create a new musician"""
|
|
slug = slugify(musician_data.name)
|
|
|
|
# Check for existing
|
|
existing = session.exec(select(Musician).where(Musician.slug == slug)).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Musician with this name already exists")
|
|
|
|
musician = Musician(
|
|
name=musician_data.name,
|
|
slug=slug,
|
|
bio=musician_data.bio,
|
|
image_url=musician_data.image_url,
|
|
primary_instrument=musician_data.primary_instrument,
|
|
)
|
|
session.add(musician)
|
|
session.commit()
|
|
session.refresh(musician)
|
|
return musician
|