122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.deps import ensure_class_access, ensure_class_permission, require_role, resolve_class_id_for_user
|
|
from app.db.database import get_db
|
|
from app.db.models import User
|
|
from app.schemas.announcement import AnnouncementCreate, AnnouncementUpdate, AnnouncementOut
|
|
from app.schemas.common import PageResponse
|
|
from app.services.announcement_service import (
|
|
create_announcement,
|
|
update_announcement,
|
|
delete_announcement,
|
|
get_announcement_by_id,
|
|
list_announcements,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/announcements", tags=["announcements"])
|
|
|
|
|
|
@router.get("/", response_model=PageResponse[AnnouncementOut])
|
|
async def get_announcements(
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
class_id: int | None = None,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
effective_class_id = resolve_class_id_for_user(user, class_id)
|
|
if effective_class_id is None:
|
|
return PageResponse(items=[], total=0, page=page, page_size=page_size, total_pages=0)
|
|
ensure_class_access(user, effective_class_id)
|
|
|
|
announcements, total = await list_announcements(db, effective_class_id, page, page_size)
|
|
total_pages = (total + page_size - 1) // page_size
|
|
|
|
items = []
|
|
for a in announcements:
|
|
items.append(
|
|
AnnouncementOut(
|
|
id=a.id,
|
|
class_id=a.class_id,
|
|
author_id=a.author_id,
|
|
author_name=a.author.name if a.author else "Unknown",
|
|
title=a.title,
|
|
content=a.content,
|
|
is_pinned=a.is_pinned,
|
|
created_at=a.created_at,
|
|
updated_at=a.updated_at,
|
|
)
|
|
)
|
|
|
|
return PageResponse(
|
|
items=items, total=total, page=page, page_size=page_size, total_pages=total_pages
|
|
)
|
|
|
|
|
|
@router.post("/", response_model=AnnouncementOut)
|
|
async def create_new_announcement(
|
|
data: AnnouncementCreate,
|
|
class_id: int | None = None,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
effective_class_id = resolve_class_id_for_user(user, class_id)
|
|
if effective_class_id is None:
|
|
raise HTTPException(status_code=400, detail="You are not assigned to a class")
|
|
ensure_class_permission(user, "announcement_manage", effective_class_id)
|
|
|
|
announcement = await create_announcement(db, effective_class_id, user.id, data)
|
|
return AnnouncementOut(
|
|
id=announcement.id,
|
|
class_id=announcement.class_id,
|
|
author_id=announcement.author_id,
|
|
author_name=user.name,
|
|
title=announcement.title,
|
|
content=announcement.content,
|
|
is_pinned=announcement.is_pinned,
|
|
created_at=announcement.created_at,
|
|
updated_at=announcement.updated_at,
|
|
)
|
|
|
|
|
|
@router.put("/{announcement_id}", response_model=AnnouncementOut)
|
|
async def update_existing_announcement(
|
|
announcement_id: int,
|
|
data: AnnouncementUpdate,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
announcement = await get_announcement_by_id(db, announcement_id)
|
|
if announcement is None:
|
|
raise HTTPException(status_code=404, detail="Announcement not found")
|
|
ensure_class_permission(user, "announcement_manage", announcement.class_id)
|
|
|
|
updated = await update_announcement(db, announcement, data)
|
|
return AnnouncementOut(
|
|
id=updated.id,
|
|
class_id=updated.class_id,
|
|
author_id=updated.author_id,
|
|
author_name=updated.author.name if updated.author else user.name,
|
|
title=updated.title,
|
|
content=updated.content,
|
|
is_pinned=updated.is_pinned,
|
|
created_at=updated.created_at,
|
|
updated_at=updated.updated_at,
|
|
)
|
|
|
|
|
|
@router.delete("/{announcement_id}")
|
|
async def delete_existing_announcement(
|
|
announcement_id: int,
|
|
user: User = Depends(require_role("super_admin", "teacher", "student")),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
announcement = await get_announcement_by_id(db, announcement_id)
|
|
if announcement is None:
|
|
raise HTTPException(status_code=404, detail="Announcement not found")
|
|
ensure_class_permission(user, "announcement_manage", announcement.class_id)
|
|
|
|
await delete_announcement(db, announcement)
|
|
return {"message": "Announcement deleted"}
|