155 lines
4.5 KiB
Python
155 lines
4.5 KiB
Python
"""Worlds router - CRUD for game worlds."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.database import get_db
|
|
from app.dependencies import CurrentUser
|
|
from app.models.world import World, WorldVisibility
|
|
|
|
router = APIRouter()
|
|
|
|
DbDep = Annotated[AsyncSession, Depends(get_db)]
|
|
|
|
|
|
@router.get("/")
|
|
async def list_worlds(
|
|
db: DbDep,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
visibility: WorldVisibility | None = None,
|
|
):
|
|
"""List worlds with optional visibility filter."""
|
|
query = select(World).options(selectinload(World.owner))
|
|
if visibility:
|
|
query = query.where(World.visibility == visibility)
|
|
query = query.offset(skip).limit(limit).order_by(World.created_at.desc())
|
|
result = await db.execute(query)
|
|
worlds = result.scalars().all()
|
|
return [
|
|
{
|
|
"id": w.id,
|
|
"name": w.name,
|
|
"description": w.description,
|
|
"visibility": w.visibility.value if w.visibility else None,
|
|
"owner_id": w.owner_id,
|
|
"created_at": w.created_at.isoformat() if w.created_at else None,
|
|
}
|
|
for w in worlds
|
|
]
|
|
|
|
|
|
@router.get("/{world_id}")
|
|
async def get_world(
|
|
world_id: int,
|
|
db: DbDep,
|
|
):
|
|
"""Get a single world by ID."""
|
|
result = await db.execute(
|
|
select(World).where(World.id == world_id).options(selectinload(World.owner))
|
|
)
|
|
world = result.scalar_one_or_none()
|
|
if not world:
|
|
raise HTTPException(status_code=404, detail="World not found")
|
|
return {
|
|
"id": world.id,
|
|
"name": world.name,
|
|
"description": world.description,
|
|
"visibility": world.visibility.value if world.visibility else None,
|
|
"owner_id": world.owner_id,
|
|
"created_at": world.created_at.isoformat() if world.created_at else None,
|
|
}
|
|
|
|
|
|
@router.post("/")
|
|
async def create_world(
|
|
world_data: dict,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
):
|
|
"""Create a new world."""
|
|
name = world_data.get("name")
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="World name is required")
|
|
|
|
world = World(
|
|
name=name,
|
|
description=world_data.get("description"),
|
|
owner_id=current_user.id,
|
|
visibility=WorldVisibility(
|
|
world_data.get("visibility", "public")
|
|
),
|
|
)
|
|
db.add(world)
|
|
await db.commit()
|
|
await db.refresh(world)
|
|
|
|
return {
|
|
"id": world.id,
|
|
"name": world.name,
|
|
"description": world.description,
|
|
"visibility": world.visibility.value,
|
|
"owner_id": world.owner_id,
|
|
"created_at": world.created_at.isoformat() if world.created_at else None,
|
|
}
|
|
|
|
|
|
@router.put("/{world_id}")
|
|
async def update_world(
|
|
world_id: int,
|
|
world_data: dict,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
):
|
|
"""Update an existing world."""
|
|
result = await db.execute(select(World).where(World.id == world_id))
|
|
world = result.scalar_one_or_none()
|
|
if not world:
|
|
raise HTTPException(status_code=404, detail="World not found")
|
|
|
|
if world.owner_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not owner of this world")
|
|
|
|
if world_data.get("name"):
|
|
world.name = world_data["name"]
|
|
if world_data.get("description") is not None:
|
|
world.description = world_data["description"]
|
|
if world_data.get("visibility"):
|
|
world.visibility = WorldVisibility(world_data["visibility"])
|
|
|
|
await db.commit()
|
|
await db.refresh(world)
|
|
|
|
return {
|
|
"id": world.id,
|
|
"name": world.name,
|
|
"description": world.description,
|
|
"visibility": world.visibility.value if world.visibility else None,
|
|
"owner_id": world.owner_id,
|
|
"created_at": world.created_at.isoformat() if world.created_at else None,
|
|
}
|
|
|
|
|
|
@router.delete("/{world_id}")
|
|
async def delete_world(
|
|
world_id: int,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
):
|
|
"""Delete a world."""
|
|
result = await db.execute(select(World).where(World.id == world_id))
|
|
world = result.scalar_one_or_none()
|
|
if not world:
|
|
raise HTTPException(status_code=404, detail="World not found")
|
|
|
|
if world.owner_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not owner of this world")
|
|
|
|
await db.delete(world)
|
|
await db.commit()
|
|
return {"detail": "World deleted"}
|