Add boss battles, level caps, and badge tracking
Introduces full boss battle system: data models (BossBattle, BossPokemon, BossResult), API endpoints for CRUD and per-run defeat tracking, and frontend UI including a sticky level cap bar with badge display on the run page, interleaved boss battle cards in the encounter list, and an admin panel section for managing boss battles and their pokemon teams. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
240
backend/src/app/api/bosses.py
Normal file
240
backend/src/app/api/bosses.py
Normal file
@@ -0,0 +1,240 @@
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.core.database import get_session
|
||||
from app.models.boss_battle import BossBattle
|
||||
from app.models.boss_pokemon import BossPokemon
|
||||
from app.models.boss_result import BossResult
|
||||
from app.models.game import Game
|
||||
from app.models.nuzlocke_run import NuzlockeRun
|
||||
from app.schemas.boss import (
|
||||
BossBattleCreate,
|
||||
BossBattleResponse,
|
||||
BossBattleUpdate,
|
||||
BossPokemonInput,
|
||||
BossResultCreate,
|
||||
BossResultResponse,
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# --- Game-scoped (admin) endpoints ---
|
||||
|
||||
|
||||
@router.get("/games/{game_id}/bosses", response_model=list[BossBattleResponse])
|
||||
async def list_bosses(
|
||||
game_id: int, session: AsyncSession = Depends(get_session)
|
||||
):
|
||||
game = await session.get(Game, game_id)
|
||||
if game is None:
|
||||
raise HTTPException(status_code=404, detail="Game not found")
|
||||
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.game_id == game_id)
|
||||
.options(selectinload(BossBattle.pokemon).selectinload(BossPokemon.pokemon))
|
||||
.order_by(BossBattle.order)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
|
||||
@router.post("/games/{game_id}/bosses", response_model=BossBattleResponse, status_code=201)
|
||||
async def create_boss(
|
||||
game_id: int,
|
||||
data: BossBattleCreate,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
game = await session.get(Game, game_id)
|
||||
if game is None:
|
||||
raise HTTPException(status_code=404, detail="Game not found")
|
||||
|
||||
boss = BossBattle(game_id=game_id, **data.model_dump())
|
||||
session.add(boss)
|
||||
await session.commit()
|
||||
|
||||
# Re-fetch with eager loading
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.id == boss.id)
|
||||
.options(selectinload(BossBattle.pokemon).selectinload(BossPokemon.pokemon))
|
||||
)
|
||||
return result.scalar_one()
|
||||
|
||||
|
||||
@router.put("/games/{game_id}/bosses/{boss_id}", response_model=BossBattleResponse)
|
||||
async def update_boss(
|
||||
game_id: int,
|
||||
boss_id: int,
|
||||
data: BossBattleUpdate,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.id == boss_id, BossBattle.game_id == game_id)
|
||||
.options(selectinload(BossBattle.pokemon).selectinload(BossPokemon.pokemon))
|
||||
)
|
||||
boss = result.scalar_one_or_none()
|
||||
if boss is None:
|
||||
raise HTTPException(status_code=404, detail="Boss battle not found")
|
||||
|
||||
for field, value in data.model_dump(exclude_unset=True).items():
|
||||
setattr(boss, field, value)
|
||||
|
||||
await session.commit()
|
||||
await session.refresh(boss)
|
||||
|
||||
# Re-fetch with eager loading
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.id == boss.id)
|
||||
.options(selectinload(BossBattle.pokemon).selectinload(BossPokemon.pokemon))
|
||||
)
|
||||
return result.scalar_one()
|
||||
|
||||
|
||||
@router.delete("/games/{game_id}/bosses/{boss_id}", status_code=204)
|
||||
async def delete_boss(
|
||||
game_id: int,
|
||||
boss_id: int,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
result = await session.execute(
|
||||
select(BossBattle).where(BossBattle.id == boss_id, BossBattle.game_id == game_id)
|
||||
)
|
||||
boss = result.scalar_one_or_none()
|
||||
if boss is None:
|
||||
raise HTTPException(status_code=404, detail="Boss battle not found")
|
||||
|
||||
await session.delete(boss)
|
||||
await session.commit()
|
||||
return Response(status_code=204)
|
||||
|
||||
|
||||
@router.put(
|
||||
"/games/{game_id}/bosses/{boss_id}/pokemon",
|
||||
response_model=BossBattleResponse,
|
||||
)
|
||||
async def set_boss_team(
|
||||
game_id: int,
|
||||
boss_id: int,
|
||||
team: list[BossPokemonInput],
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.id == boss_id, BossBattle.game_id == game_id)
|
||||
.options(selectinload(BossBattle.pokemon))
|
||||
)
|
||||
boss = result.scalar_one_or_none()
|
||||
if boss is None:
|
||||
raise HTTPException(status_code=404, detail="Boss battle not found")
|
||||
|
||||
# Remove existing team
|
||||
for p in boss.pokemon:
|
||||
await session.delete(p)
|
||||
|
||||
# Add new team
|
||||
for item in team:
|
||||
bp = BossPokemon(
|
||||
boss_battle_id=boss_id,
|
||||
pokemon_id=item.pokemon_id,
|
||||
level=item.level,
|
||||
order=item.order,
|
||||
)
|
||||
session.add(bp)
|
||||
|
||||
await session.commit()
|
||||
|
||||
# Re-fetch with eager loading
|
||||
result = await session.execute(
|
||||
select(BossBattle)
|
||||
.where(BossBattle.id == boss.id)
|
||||
.options(selectinload(BossBattle.pokemon).selectinload(BossPokemon.pokemon))
|
||||
)
|
||||
return result.scalar_one()
|
||||
|
||||
|
||||
# --- Run-scoped endpoints ---
|
||||
|
||||
|
||||
@router.get("/runs/{run_id}/boss-results", response_model=list[BossResultResponse])
|
||||
async def list_boss_results(
|
||||
run_id: int, session: AsyncSession = Depends(get_session)
|
||||
):
|
||||
run = await session.get(NuzlockeRun, run_id)
|
||||
if run is None:
|
||||
raise HTTPException(status_code=404, detail="Run not found")
|
||||
|
||||
result = await session.execute(
|
||||
select(BossResult)
|
||||
.where(BossResult.run_id == run_id)
|
||||
.order_by(BossResult.id)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
|
||||
@router.post("/runs/{run_id}/boss-results", response_model=BossResultResponse, status_code=201)
|
||||
async def create_boss_result(
|
||||
run_id: int,
|
||||
data: BossResultCreate,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
run = await session.get(NuzlockeRun, run_id)
|
||||
if run is None:
|
||||
raise HTTPException(status_code=404, detail="Run not found")
|
||||
|
||||
boss = await session.get(BossBattle, data.boss_battle_id)
|
||||
if boss is None:
|
||||
raise HTTPException(status_code=404, detail="Boss battle not found")
|
||||
|
||||
# Check for existing result (upsert)
|
||||
existing = await session.execute(
|
||||
select(BossResult).where(
|
||||
BossResult.run_id == run_id,
|
||||
BossResult.boss_battle_id == data.boss_battle_id,
|
||||
)
|
||||
)
|
||||
result = existing.scalar_one_or_none()
|
||||
|
||||
if result:
|
||||
result.result = data.result
|
||||
result.attempts = data.attempts
|
||||
result.completed_at = datetime.now(timezone.utc) if data.result == "won" else None
|
||||
else:
|
||||
result = BossResult(
|
||||
run_id=run_id,
|
||||
boss_battle_id=data.boss_battle_id,
|
||||
result=data.result,
|
||||
attempts=data.attempts,
|
||||
completed_at=datetime.now(timezone.utc) if data.result == "won" else None,
|
||||
)
|
||||
session.add(result)
|
||||
|
||||
await session.commit()
|
||||
await session.refresh(result)
|
||||
return result
|
||||
|
||||
|
||||
@router.delete("/runs/{run_id}/boss-results/{result_id}", status_code=204)
|
||||
async def delete_boss_result(
|
||||
run_id: int,
|
||||
result_id: int,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
result = await session.execute(
|
||||
select(BossResult).where(
|
||||
BossResult.id == result_id, BossResult.run_id == run_id
|
||||
)
|
||||
)
|
||||
boss_result = result.scalar_one_or_none()
|
||||
if boss_result is None:
|
||||
raise HTTPException(status_code=404, detail="Boss result not found")
|
||||
|
||||
await session.delete(boss_result)
|
||||
await session.commit()
|
||||
return Response(status_code=204)
|
||||
Reference in New Issue
Block a user