Spaces:
Running
Running
File size: 5,947 Bytes
89ae94f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 |
from fastapi import APIRouter, Depends, HTTPException, status, Body, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional
import logging
from src.api.database import get_db
from src.api.auth import get_current_user
from src.api.schemas import (
Threat, ThreatCreate, ThreatUpdate, ThreatFilter,
PaginationParams, User
)
from src.api.services.threat_service import (
create_threat, get_threat_by_id, update_threat,
delete_threat, get_threats, get_threat_statistics
)
# Configure logger
logger = logging.getLogger(__name__)
router = APIRouter(
tags=["threats"],
responses={404: {"description": "Not found"}}
)
@router.post("/", response_model=Threat, status_code=status.HTTP_201_CREATED)
async def create_threat_endpoint(
threat_data: ThreatCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Create a new threat.
Args:
threat_data: Threat data
db: Database session
current_user: Current authenticated user
Returns:
Threat: Created threat
"""
try:
threat = await create_threat(db, threat_data)
if not threat:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Failed to create threat"
)
return threat
except Exception as e:
logger.error(f"Error creating threat: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"An error occurred: {str(e)}"
)
@router.get("/{threat_id}", response_model=Threat)
async def get_threat_endpoint(
threat_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get threat by ID.
Args:
threat_id: Threat ID
db: Database session
current_user: Current authenticated user
Returns:
Threat: Threat data
"""
threat = await get_threat_by_id(db, threat_id)
if not threat:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Threat with ID {threat_id} not found"
)
return threat
@router.put("/{threat_id}", response_model=Threat)
async def update_threat_endpoint(
threat_id: int,
threat_data: ThreatUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Update threat.
Args:
threat_id: Threat ID
threat_data: Threat data
db: Database session
current_user: Current authenticated user
Returns:
Threat: Updated threat
"""
# Check if threat exists
threat = await get_threat_by_id(db, threat_id)
if not threat:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Threat with ID {threat_id} not found"
)
# Update threat
updated_threat = await update_threat(db, threat_id, threat_data)
if not updated_threat:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Failed to update threat"
)
return updated_threat
@router.delete("/{threat_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_threat_endpoint(
threat_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Delete threat.
Args:
threat_id: Threat ID
db: Database session
current_user: Current authenticated user
"""
# Check if threat exists
threat = await get_threat_by_id(db, threat_id)
if not threat:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Threat with ID {threat_id} not found"
)
# Delete threat
deleted = await delete_threat(db, threat_id)
if not deleted:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Failed to delete threat"
)
@router.get("/", response_model=List[Threat])
async def get_threats_endpoint(
pagination: PaginationParams = Depends(),
severity: Optional[List[str]] = Query(None),
status: Optional[List[str]] = Query(None),
category: Optional[List[str]] = Query(None),
search: Optional[str] = Query(None),
from_date: Optional[str] = Query(None),
to_date: Optional[str] = Query(None),
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get threats with filtering and pagination.
Args:
pagination: Pagination parameters
severity: Filter by severity
status: Filter by status
category: Filter by category
search: Search in title and description
from_date: Filter from date
to_date: Filter to date
db: Database session
current_user: Current authenticated user
Returns:
List[Threat]: List of threats
"""
# Create filter params
filter_params = ThreatFilter(
severity=severity,
status=status,
category=category,
search=search,
from_date=from_date,
to_date=to_date
)
# Get threats
threats, total = await get_threats(db, filter_params, pagination)
return threats
@router.get("/statistics", response_model=dict)
async def get_threat_statistics_endpoint(
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get threat statistics.
Args:
db: Database session
current_user: Current authenticated user
Returns:
dict: Threat statistics
"""
statistics = await get_threat_statistics(db)
return statistics |