94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
from ninja import Router
|
|
|
|
from apps.tickets.models import Ticket, TicketEvent
|
|
from apps.tickets.schemas import TicketIn, TicketOut, TicketStatusIn, TicketUpdateIn
|
|
from apps.tickets.services import transition_ticket
|
|
|
|
router = Router(tags=["tickets"])
|
|
|
|
|
|
def next_ticket_no() -> str:
|
|
prefix = timezone.localdate().strftime("TK%Y%m%d")
|
|
count = Ticket.objects.filter(ticket_no__startswith=prefix).count() + 1
|
|
return f"{prefix}{count:04d}"
|
|
|
|
|
|
@router.post("", response=TicketOut)
|
|
def create_ticket(request, payload: TicketIn):
|
|
assignee = None
|
|
if payload.assignee_id:
|
|
assignee = get_user_model().objects.filter(id=payload.assignee_id).first()
|
|
ticket = Ticket.objects.create(
|
|
ticket_no=next_ticket_no(),
|
|
title=payload.title,
|
|
description=payload.description,
|
|
priority=payload.priority,
|
|
impact_scope=payload.impact_scope,
|
|
environment=payload.environment,
|
|
related_system=payload.related_system,
|
|
assignee=assignee,
|
|
created_by=request.user if request.user.is_authenticated else None,
|
|
)
|
|
TicketEvent.objects.create(
|
|
ticket=ticket,
|
|
event_type="created",
|
|
to_value={"status": ticket.status},
|
|
)
|
|
return ticket
|
|
|
|
|
|
@router.get("", response=list[TicketOut])
|
|
def list_tickets(request, status: str | None = None, q: str | None = None):
|
|
queryset = Ticket.objects.all()
|
|
if status:
|
|
queryset = queryset.filter(status=status)
|
|
if q:
|
|
queryset = queryset.filter(Q(ticket_no__icontains=q) | Q(title__icontains=q))
|
|
return queryset[:100]
|
|
|
|
|
|
@router.get("/{ticket_id}", response=TicketOut)
|
|
def get_ticket(request, ticket_id: int):
|
|
return Ticket.objects.get(id=ticket_id)
|
|
|
|
|
|
@router.patch("/{ticket_id}", response=TicketOut)
|
|
def update_ticket(request, ticket_id: int, payload: TicketUpdateIn):
|
|
ticket = Ticket.objects.get(id=ticket_id)
|
|
before = {
|
|
"title": ticket.title,
|
|
"priority": ticket.priority,
|
|
"impact_scope": ticket.impact_scope,
|
|
"environment": ticket.environment,
|
|
"related_system": ticket.related_system,
|
|
}
|
|
for field, value in payload.dict(exclude_unset=True).items():
|
|
if field == "assignee_id":
|
|
user = get_user_model().objects.filter(id=value).first() if value else None
|
|
ticket.assignee = user
|
|
else:
|
|
setattr(ticket, field, value)
|
|
ticket.save()
|
|
TicketEvent.objects.create(
|
|
ticket=ticket,
|
|
event_type="updated",
|
|
from_value=before,
|
|
to_value=payload.dict(exclude_unset=True),
|
|
operator=request.user if request.user.is_authenticated else None,
|
|
)
|
|
return ticket
|
|
|
|
|
|
@router.post("/{ticket_id}/status", response=TicketOut)
|
|
def update_ticket_status(request, ticket_id: int, payload: TicketStatusIn):
|
|
ticket = Ticket.objects.get(id=ticket_id)
|
|
return transition_ticket(
|
|
ticket,
|
|
payload.status,
|
|
payload.remark,
|
|
request.user if request.user.is_authenticated else None,
|
|
)
|