85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.auth import get_current_admin_user
|
|
from app.database import get_session
|
|
from app.models import Program
|
|
from app.schemas import ProgramCreate, ProgramUpdate, ProgramResponse
|
|
from app.user_models import User
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _day_filter(day: int):
|
|
"""Return the SQLAlchemy filter for a given day_of_week."""
|
|
return Program.day_of_week == day
|
|
|
|
|
|
@router.get("/", response_model=list[ProgramResponse])
|
|
async def list_programs(
|
|
day: Optional[int] = Query(None, description="Filter by day_of_week (1=Mon … 7=Sun)"),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
query = select(Program).order_by(Program.day_of_week, Program.time)
|
|
if day is not None:
|
|
query = query.filter(_day_filter(day))
|
|
result = await session.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.get("/{program_id}", response_model=ProgramResponse)
|
|
async def get_program(program_id: int, session: AsyncSession = Depends(get_session)):
|
|
result = await session.execute(select(Program).where(Program.id == program_id))
|
|
program = result.scalar_one_or_none()
|
|
if not program:
|
|
return {"error": "Program not found"}
|
|
return program
|
|
|
|
|
|
@router.post("/", response_model=ProgramResponse, status_code=201)
|
|
async def create_program(
|
|
payload: ProgramCreate,
|
|
session: AsyncSession = Depends(get_session),
|
|
_: User = Depends(get_current_admin_user),
|
|
):
|
|
program = Program(**payload.model_dump())
|
|
session.add(program)
|
|
await session.commit()
|
|
await session.refresh(program)
|
|
return program
|
|
|
|
|
|
@router.put("/{program_id}", response_model=ProgramResponse)
|
|
async def update_program(
|
|
program_id: int,
|
|
payload: ProgramUpdate,
|
|
session: AsyncSession = Depends(get_session),
|
|
_: User = Depends(get_current_admin_user),
|
|
):
|
|
result = await session.execute(select(Program).where(Program.id == program_id))
|
|
program = result.scalar_one_or_none()
|
|
if not program:
|
|
return {"error": "Program not found"}
|
|
for key, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(program, key, value)
|
|
await session.commit()
|
|
await session.refresh(program)
|
|
return program
|
|
|
|
|
|
@router.delete("/{program_id}", status_code=204)
|
|
async def delete_program(
|
|
program_id: int,
|
|
session: AsyncSession = Depends(get_session),
|
|
_: User = Depends(get_current_admin_user),
|
|
):
|
|
result = await session.execute(select(Program).where(Program.id == program_id))
|
|
program = result.scalar_one_or_none()
|
|
if not program:
|
|
return {"error": "Program not found"}
|
|
await session.delete(program)
|
|
await session.commit()
|