144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from datetime import datetime
|
|
from config.database import get_db
|
|
from models.information import Information
|
|
from models.user import User
|
|
from schemas.information import InformationCreate, InformationUpdate, InformationResponse
|
|
from utils.security import get_current_active_user, get_admin_user
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/", response_model=List[InformationResponse])
|
|
async def get_informations(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user),
|
|
category: str = None,
|
|
published_only: bool = True
|
|
):
|
|
"""Get all informations, optionally filtered by category and published status."""
|
|
query = db.query(Information)
|
|
|
|
if published_only:
|
|
query = query.filter(Information.is_published == True)
|
|
|
|
if category:
|
|
query = query.filter(Information.category == category)
|
|
|
|
informations = query.order_by(Information.priority.desc(), Information.created_at.desc()).all()
|
|
return informations
|
|
|
|
@router.get("/public", response_model=List[InformationResponse])
|
|
async def get_public_informations(
|
|
db: Session = Depends(get_db),
|
|
category: str = None
|
|
):
|
|
"""Get public informations without authentication."""
|
|
query = db.query(Information).filter(Information.is_published == True)
|
|
|
|
if category:
|
|
query = query.filter(Information.category == category)
|
|
|
|
informations = query.order_by(Information.priority.desc(), Information.created_at.desc()).all()
|
|
return informations
|
|
|
|
@router.get("/{information_id}", response_model=InformationResponse)
|
|
async def get_information(
|
|
information_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""Get a specific information."""
|
|
information = db.query(Information).filter(Information.id == information_id).first()
|
|
if not information:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Information not found"
|
|
)
|
|
|
|
if not information.is_published and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Information not found"
|
|
)
|
|
|
|
return information
|
|
|
|
@router.post("/", response_model=InformationResponse)
|
|
async def create_information(
|
|
information_data: InformationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_admin_user)
|
|
):
|
|
"""Create a new information (admin only)."""
|
|
information = Information(**information_data.dict())
|
|
db.add(information)
|
|
db.commit()
|
|
db.refresh(information)
|
|
return information
|
|
|
|
@router.put("/{information_id}", response_model=InformationResponse)
|
|
async def update_information(
|
|
information_id: int,
|
|
information_update: InformationUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_admin_user)
|
|
):
|
|
"""Update an information (admin only)."""
|
|
information = db.query(Information).filter(Information.id == information_id).first()
|
|
if not information:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Information not found"
|
|
)
|
|
|
|
for field, value in information_update.dict(exclude_unset=True).items():
|
|
setattr(information, field, value)
|
|
|
|
information.updated_at = datetime.utcnow()
|
|
db.commit()
|
|
db.refresh(information)
|
|
return information
|
|
|
|
@router.delete("/{information_id}")
|
|
async def delete_information(
|
|
information_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_admin_user)
|
|
):
|
|
"""Delete an information (admin only)."""
|
|
information = db.query(Information).filter(Information.id == information_id).first()
|
|
if not information:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Information not found"
|
|
)
|
|
|
|
db.delete(information)
|
|
db.commit()
|
|
return {"message": "Information deleted successfully"}
|
|
|
|
@router.put("/{information_id}/toggle-publish")
|
|
async def toggle_information_publish(
|
|
information_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_admin_user)
|
|
):
|
|
"""Toggle information published status (admin only)."""
|
|
information = db.query(Information).filter(Information.id == information_id).first()
|
|
if not information:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Information not found"
|
|
)
|
|
|
|
information.is_published = not information.is_published
|
|
information.updated_at = datetime.utcnow()
|
|
db.commit()
|
|
|
|
return {
|
|
"message": f"Information {'published' if information.is_published else 'unpublished'} successfully",
|
|
"is_published": information.is_published
|
|
}
|