75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
import uuid
|
|
from sqlalchemy.exc import NoResultFound
|
|
|
|
from app.core.database import get_db
|
|
from app.models.ticket import Ticket # Asegúrate de tener el modelo Ticket
|
|
from app.models.category import Category # Ensure Category model is imported
|
|
from app.api import deps # Para autenticación y permisos
|
|
from app.models.user import User # Asegúrate de importar el modelo User
|
|
|
|
router = APIRouter()
|
|
|
|
# Esquema para crear un ticket
|
|
class TicketCreate(BaseModel):
|
|
title: str
|
|
description: str
|
|
category_id: uuid.UUID
|
|
priority: str
|
|
|
|
# Esquema para devolver un ticket
|
|
class TicketResponse(BaseModel):
|
|
id: uuid.UUID
|
|
title: str
|
|
description: str
|
|
category_id: uuid.UUID
|
|
priority: str
|
|
created_at: str
|
|
|
|
@router.post("/", response_model=TicketResponse)
|
|
async def create_ticket(
|
|
ticket: TicketCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_active_user) # Verifica que el usuario esté autenticado
|
|
):
|
|
# Validate if the category exists
|
|
try:
|
|
category = await db.execute(select(Category).where(Category.id == ticket.category_id))
|
|
category = category.scalar_one()
|
|
except NoResultFound:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
|
|
# Crear un nuevo ticket
|
|
db_ticket = Ticket(
|
|
ticket_number=str(uuid.uuid4())[:8], # Generate a unique ticket number
|
|
subject=ticket.title,
|
|
description=ticket.description,
|
|
category_id=ticket.category_id,
|
|
priority=ticket.priority,
|
|
created_by=current_user.id,
|
|
tenant_id=current_user.tenant_id
|
|
)
|
|
db.add(db_ticket)
|
|
await db.commit()
|
|
await db.refresh(db_ticket)
|
|
return db_ticket
|
|
|
|
@router.get("/", response_model=List[TicketResponse])
|
|
async def get_tickets(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_active_user) # Verifica que el usuario esté autenticado
|
|
):
|
|
try:
|
|
result = await db.execute(select(Ticket).where(Ticket.tenant_id == current_user.tenant_id))
|
|
tickets = result.scalars().all()
|
|
return tickets
|
|
except Exception as e:
|
|
# Log the error for debugging purposes
|
|
import logging
|
|
logger = logging.getLogger("tickets")
|
|
logger.exception("Error fetching tickets for tenant_id=%s", current_user.tenant_id)
|
|
raise HTTPException(status_code=500, detail="An error occurred while fetching tickets.") |