from sqlalchemy.orm import Session, joinedload from fastapi import HTTPException, status, Request, BackgroundTasks, UploadFile, File from models.project_model import Project from models.ticket_model import Ticket from models.user_model import User from models.label_model import TicketLabel, Release, TicketRelease from schema.ticket_schema import TicketCreateRequestSchema from services.activity_log_service import log_activity from utils.logging import logger from uuid import UUID from typing import Optional, List import csv import json from backgroundtask.add_tickets_from_csv import process_tickets import asyncio from functools import partial import io from routers.websocket_router import manager def create_ticket_for_project(request: Request, payload: TicketCreateRequestSchema, background_tasks: BackgroundTasks, db: Session): try: user = request.state.user logger.info(f"User {user.id} creating ticket for project {payload.project_id}") # Check if project exists and belongs to the user's organization project = db.query(Project).filter(Project.id == payload.project_id, Project.org_id == user.org_id).first() if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Project not found or not in user's organization" ) # Increment ticket counter in a safe transaction context project.ticket_counter += 1 db.add(project) # Format ticket key number ticket_number = f"{project.project_key}-{project.ticket_counter}" # Initialize new ticket new_ticket = Ticket( project_id=payload.project_id, reporter_id=user.id, ticket_number=ticket_number, title=payload.title, description=payload.description, type=payload.type, status=payload.status, priority=payload.priority, story_points=payload.story_points, due_date=payload.due_date ) if payload.assignee_ids: new_ticket.assignee_ids = payload.assignee_ids elif payload.assignee_id: new_ticket.assignee_ids = [payload.assignee_id] db.add(new_ticket) db.flush() # Save label associations if payload.label_ids: for lid in payload.label_ids: ticket_label = TicketLabel(ticket_id=new_ticket.id, label_id=lid) db.add(ticket_label) # Save releases if payload.release_names: for rel_name in payload.release_names: stripped_name = rel_name.strip() if stripped_name: # Check if release exists in organization release = ( db.query(Release) .filter(Release.org_id == user.org_id, Release.name == stripped_name) .first() ) if not release: release = Release(org_id=user.org_id, name=stripped_name) db.add(release) db.flush() # to get release.id # Create the link ticket_release = TicketRelease(ticket_id=new_ticket.id, release_id=release.id) db.add(ticket_release) db.commit() # Create notifications and send emails for initial assignees if new_ticket.assignee_ids: from services.notification_service import create_notification from utils.send_assignment_email import send_assignment_email ticket_link = f"/dashboard/project/{project.id}?ticket={new_ticket.ticket_number}" for assignee_id in new_ticket.assignee_ids: if str(assignee_id) != str(user.id): create_notification( db=db, user_id=assignee_id, message=f"{user.name} assigned you to ticket {new_ticket.ticket_number}", link=ticket_link ) assigned_user = db.query(User).filter(User.id == assignee_id).first() if assigned_user: frontend_url = __import__("os").getenv("FRONTEND_URL", "http://localhost:3000") full_link = f"{frontend_url}{ticket_link}" background_tasks.add_task( send_assignment_email, email=assigned_user.email, assigned_by_name=user.name, ticket_name=new_ticket.title, ticket_number=new_ticket.ticket_number, ticket_link=full_link ) # Eagerly load the newly created ticket with all its relations new_ticket = ( db.query(Ticket) .filter(Ticket.id == new_ticket.id) .options( joinedload(Ticket.ticket_labels).joinedload(TicketLabel.label), joinedload(Ticket.releases) ) .first() ) background_tasks.add_task( manager.broadcast_to_project, str(payload.project_id), {"event": "ticket_created", "data": {"id": str(new_ticket.id)}} ) return new_ticket except HTTPException as e: raise e except Exception as e: db.rollback() logger.error(f"Error creating ticket: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to create ticket: {str(e)}" ) def query_filtered_tickets( db: Session, org_id: UUID, project_id: Optional[UUID] = None, limit: int = 20, offset: int = 0, assignee_id: Optional[List[str]] = None, reporter_id: Optional[List[str]] = None, priority: Optional[List[str]] = None, type: Optional[List[str]] = None, status: Optional[List[str]] = None, label_id: Optional[List[str]] = None, release_name: Optional[List[str]] = None, created_after: Optional[str] = None, created_before: Optional[str] = None, q: Optional[str] = None, sort_by: Optional[str] = "newest" ): query = db.query(Ticket).options( joinedload(Ticket.ticket_labels).joinedload(TicketLabel.label), joinedload(Ticket.releases) ) if project_id: query = query.filter(Ticket.project_id == project_id) else: # Join project to ensure org filter query = query.join(Project).filter(Project.org_id == org_id) # Apply filters if q: search_pattern = f"%{q}%" query = query.filter( (Ticket.title.ilike(search_pattern)) | (Ticket.ticket_number.ilike(search_pattern)) ) if assignee_id: try: assignee_uuids = [UUID(a) for a in assignee_id if a] if assignee_uuids: query = query.filter(Ticket.assignee_ids.overlap(assignee_uuids)) except ValueError: pass if reporter_id: try: reporter_uuids = [UUID(r) for r in reporter_id if r] if reporter_uuids: query = query.filter(Ticket.reporter_id.in_(reporter_uuids)) except ValueError: pass if priority: query = query.filter(Ticket.priority.in_(priority)) if type: query = query.filter(Ticket.type.in_(type)) if status: query = query.filter(Ticket.status.in_(status)) if label_id: try: label_uuids = [UUID(l) for l in label_id if l] if label_uuids: query = query.join(Ticket.ticket_labels).filter(TicketLabel.label_id.in_(label_uuids)) except ValueError: pass if release_name: query = query.join(Ticket.ticket_releases).join(Release).filter(Release.name.in_(release_name)) if created_after: try: from datetime import datetime query = query.filter(Ticket.created_at >= datetime.fromisoformat(created_after)) except ValueError: pass if created_before: try: from datetime import datetime query = query.filter(Ticket.created_at <= datetime.fromisoformat(created_before)) except ValueError: pass # Handle sorting if sort_by == "newest": query = query.order_by(Ticket.created_at.desc()) elif sort_by == "oldest": query = query.order_by(Ticket.created_at.asc()) elif sort_by == "pipeline_old_to_new": query = query.order_by(Ticket.updated_at.asc()) elif sort_by == "pipeline_new_to_old": query = query.order_by(Ticket.updated_at.desc()) elif sort_by == "estimate_high_low": query = query.order_by(Ticket.story_points.desc()) elif sort_by == "estimate_low_high": query = query.order_by(Ticket.story_points.asc()) elif sort_by == "title_a_z": query = query.order_by(Ticket.title.asc()) elif sort_by == "title_z_a": query = query.order_by(Ticket.title.desc()) else: # Default manual sort query = query.order_by(Ticket.ticket_number.asc()) return query.distinct().offset(offset).limit(limit).all() def get_tickets_by_project_id( project_id: UUID, request: Request, db: Session, limit: int = 20, offset: int = 0, assignee_id: Optional[List[str]] = None, reporter_id: Optional[List[str]] = None, priority: Optional[List[str]] = None, type: Optional[List[str]] = None, status: Optional[List[str]] = None, label_id: Optional[List[str]] = None, release_name: Optional[List[str]] = None, created_after: Optional[str] = None, created_before: Optional[str] = None, q: Optional[str] = None, sort_by: Optional[str] = "manual" ): try: user = request.state.user # Verify project is in the user's organization project = db.query(Project).filter(Project.id == project_id, Project.org_id == user.org_id).first() if not project: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Project not found or not in user's organization" ) return query_filtered_tickets( db=db, org_id=user.org_id, project_id=project_id, limit=limit, offset=offset, assignee_id=assignee_id, reporter_id=reporter_id, priority=priority, type=type, status=status, label_id=label_id, release_name=release_name, created_after=created_after, created_before=created_before, q=q, sort_by=sort_by ) except HTTPException as e: raise e except Exception as e: logger.error(f"Error getting tickets: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get tickets: {str(e)}" ) def get_all_organization_tickets_from_db( request: Request, db: Session, limit: int = 20, offset: int = 0, assignee_id: Optional[List[str]] = None, reporter_id: Optional[List[str]] = None, priority: Optional[List[str]] = None, type: Optional[List[str]] = None, status: Optional[List[str]] = None, label_id: Optional[List[str]] = None, release_name: Optional[List[str]] = None, created_after: Optional[str] = None, created_before: Optional[str] = None, q: Optional[str] = None, sort_by: Optional[str] = "manual" ): try: user = request.state.user if not user or not user.org_id: return [] return query_filtered_tickets( db=db, org_id=user.org_id, project_id=None, limit=limit, offset=offset, assignee_id=assignee_id, reporter_id=reporter_id, priority=priority, type=type, status=status, label_id=label_id, release_name=release_name, created_after=created_after, created_before=created_before, q=q, sort_by=sort_by ) except Exception as e: logger.error(f"Error fetching all organization tickets: {str(e)}") return [] def update_ticket_by_id(ticket_id: UUID, request: Request, payload: any, background_tasks: BackgroundTasks, db: Session): try: user = request.state.user logger.info(f"User {user.id} updating ticket {ticket_id}") ticket = db.query(Ticket).filter(Ticket.id == ticket_id).first() if not ticket: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ticket not found" ) project = db.query(Project).filter(Project.id == ticket.project_id, Project.org_id == user.org_id).first() if not project: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied: project organization mismatch" ) update_data = payload.dict(exclude_unset=True) # ── Track field changes for activity log ── tracked_fields = ["status", "priority", "type", "title", "story_points", "due_date"] for field in tracked_fields: old_raw = getattr(ticket, field, None) new_raw = update_data.get(field) # Extract .value from Python enums to get clean strings like "MEDIUM" instead of "TicketPriority.MEDIUM" old_val = old_raw.value if hasattr(old_raw, "value") else str(old_raw or "") new_val = new_raw.value if hasattr(new_raw, "value") else str(new_raw or "") if field in update_data and old_val != new_val: log_activity( db=db, user_id=user.id, ticket_id=ticket.id, action=f"changed {field.replace('_', ' ')}", old_value=old_val, new_value=new_val, ) # Track assignee changes old_assignees = set(str(a) for a in (ticket.assignee_ids or [])) new_assignees_raw = update_data.get("assignee_ids") or ( [update_data["assignee_id"]] if update_data.get("assignee_id") else None ) if new_assignees_raw is not None: new_assignees = set(str(a) for a in new_assignees_raw if a) if old_assignees != new_assignees: log_activity( db=db, user_id=user.id, ticket_id=ticket.id, action="changed assignees", old_value=", ".join(sorted(old_assignees)) if old_assignees else "none", new_value=", ".join(sorted(new_assignees)) if new_assignees else "none", ) # Create notifications and send emails for newly added assignees added_assignees = new_assignees - old_assignees if added_assignees: from services.notification_service import create_notification from utils.send_assignment_email import send_assignment_email ticket_link = f"/dashboard/project/{ticket.project_id}?ticket={ticket.ticket_number}" for assignee_id in added_assignees: if assignee_id != str(user.id): create_notification( db=db, user_id=UUID(assignee_id), message=f"{user.name} assigned you to ticket {ticket.ticket_number}", link=ticket_link ) assigned_user = db.query(User).filter(User.id == assignee_id).first() if assigned_user: frontend_url = __import__("os").getenv("FRONTEND_URL", "http://localhost:3000") full_link = f"{frontend_url}{ticket_link}" background_tasks.add_task( send_assignment_email, email=assigned_user.email, assigned_by_name=user.name, ticket_name=update_data.get("title", ticket.title), ticket_number=ticket.ticket_number, ticket_link=full_link ) # Handle assignee_ids and assignee_id safe transition if "assignee_ids" in update_data: ticket.assignee_ids = update_data["assignee_ids"] or [] del update_data["assignee_ids"] if "assignee_id" in update_data: del update_data["assignee_id"] elif "assignee_id" in update_data: val = update_data["assignee_id"] ticket.assignee_ids = [val] if val else [] del update_data["assignee_id"] for key, value in update_data.items(): setattr(ticket, key, value) db.commit() db.refresh(ticket) background_tasks.add_task( manager.broadcast_to_project, str(ticket.project_id), {"event": "ticket_updated", "data": {"id": str(ticket.id)}} ) return ticket except HTTPException as e: raise e except Exception as e: db.rollback() logger.error(f"Error updating ticket: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update ticket: {str(e)}" ) def delete_ticket_by_id(ticket_id: UUID, request: Request, background_tasks: BackgroundTasks, db: Session): try: user = request.state.user logger.info(f"User {user.id} deleting ticket {ticket_id}") ticket = db.query(Ticket).filter(Ticket.id == ticket_id).first() if not ticket: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ticket not found" ) project = db.query(Project).filter(Project.id == ticket.project_id, Project.org_id == user.org_id).first() if not project: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) ticket_data = { "id": ticket.id, "project_id": ticket.project_id, "parent_ticket_id": ticket.parent_ticket_id, "assignee_id": ticket.assignee_id, "reporter_id": ticket.reporter_id, "ticket_number": ticket.ticket_number, "title": ticket.title, "description": ticket.description, "type": ticket.type, "status": ticket.status, "priority": ticket.priority, "story_points": ticket.story_points, "due_date": ticket.due_date, "message": "Ticket deleted successfully" } db.delete(ticket) db.commit() background_tasks.add_task( manager.broadcast_to_project, str(ticket.project_id), {"event": "ticket_deleted", "data": {"id": str(ticket.id)}} ) return ticket_data except HTTPException as e: raise e except Exception as e: db.rollback() logger.error(f"Error deleting ticket: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete ticket: {str(e)}" ) def get_ticket_by_number(ticket_number: str, request: Request, db: Session): try: user = request.state.user ticket = db.query(Ticket).options( joinedload(Ticket.assignee), joinedload(Ticket.reporter) ).filter(Ticket.ticket_number == ticket_number, Ticket.org_id == user.org_id).first() if not ticket: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ticket not found or not in user's organization" ) # Manually attach assignee_ids for the response schema if ticket.assignee: ticket.assignee_ids = [str(ticket.assignee.id)] else: ticket.assignee_ids = [] # Ensure project access project = db.query(Project).filter(Project.id == ticket.project_id).first() if not project or project.org_id != user.org_id: raise HTTPException(status_code=403, detail="Not authorized to view this ticket's project") return ticket except HTTPException: raise except Exception as e: logger.error(f"Error fetching ticket {ticket_number}: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch ticket" ) async def upload_tickets(request: Request, background_tasks: BackgroundTasks, db: Session, file: UploadFile = File(...)): try: loop = asyncio.get_event_loop() # Read file contents first (async) contents = await file.read() def parse_csv(content_bytes): content_str = content_bytes.decode('utf-8', errors='ignore') reader = csv.DictReader(io.StringIO(content_str)) tickets_list = [] for row in reader: ticket_dict = {} for k, v in row.items(): if v == "": ticket_dict[k] = None else: ticket_dict[k] = v tickets_list.append(ticket_dict) return tickets_list # Offload blocking csv ops to thread pool tickets = await loop.run_in_executor(None, parse_csv, contents) logger.info(f"Tickets parsed: {len(tickets)}") background_tasks.add_task(process_tickets, tickets, "a1b2c3d4-e5f6-7890-abcd-ef1234567890", "c15a8014-ea9c-406c-b04d-0491b4c5af5d") return {"message": "Request submitted successfully!"} except HTTPException as e: raise except Exception as e: logger.error(f"Error Uploading ticket: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to Upload ticket" )