75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
from sqlalchemy import insert, select, update, delete
|
|
from fastapi import HTTPException
|
|
from models.db import technical_issues_table
|
|
from models.schemas import CreateTechnicalIssue, UpdateTechnicalIssue
|
|
from services.auth_service import AuthService
|
|
|
|
|
|
class TechnicalIssueService:
|
|
@staticmethod
|
|
async def create_issue(issue: CreateTechnicalIssue, db, token: str):
|
|
user = await AuthService.get_current_user(token, db);
|
|
query = insert(technical_issues_table).values(
|
|
user_id=user["id"],
|
|
description=issue.description,
|
|
status=issue.status
|
|
)
|
|
try:
|
|
result = await db.execute(query)
|
|
await db.commit()
|
|
issue_id = result.inserted_primary_key[0]
|
|
return {"id": issue_id, **issue.model_dump()}
|
|
except Exception as e:
|
|
await db.rollback()
|
|
raise HTTPException(status_code=500, detail=f"Could not create issue: {str(e)}")
|
|
|
|
@staticmethod
|
|
async def get_issue(issue_id: int, db, token: str):
|
|
await AuthService.admin_required(token, db)
|
|
query = select(technical_issues_table).where(technical_issues_table.c.id == issue_id)
|
|
result = await db.execute(query)
|
|
issue = result.mappings().fetchone()
|
|
if not issue:
|
|
raise HTTPException(status_code=404, detail="Technical issue not found")
|
|
return dict(issue)
|
|
|
|
@staticmethod
|
|
async def get_all_issues(db, token: str):
|
|
await AuthService.admin_required(token, db)
|
|
query = select(technical_issues_table)
|
|
result = await db.execute(query)
|
|
technical_issues = result.mappings().all()
|
|
return [dict(technical_issue) for technical_issue in technical_issues]
|
|
|
|
@staticmethod
|
|
async def update_issue(issue_id: int, issue_data: UpdateTechnicalIssue, db, token: str):
|
|
await AuthService.admin_required(token, db)
|
|
query = (
|
|
update(technical_issues_table)
|
|
.where(technical_issues_table.c.id == issue_id)
|
|
.values(**issue_data.model_dump(exclude_unset=True))
|
|
)
|
|
try:
|
|
result = await db.execute(query)
|
|
if result.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Technical issue not found")
|
|
await db.commit()
|
|
return {"message": "Technical issue updated successfully"}
|
|
except Exception as e:
|
|
await db.rollback()
|
|
raise HTTPException(status_code=500, detail=f"Could not update issue: {str(e)}")
|
|
|
|
@staticmethod
|
|
async def delete_issue(issue_id: int, db, token: str):
|
|
await AuthService.admin_required(token, db)
|
|
query = delete(technical_issues_table).where(technical_issues_table.c.id == issue_id)
|
|
try:
|
|
result = await db.execute(query)
|
|
if result.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Technical issue not found")
|
|
await db.commit()
|
|
return {"message": "Technical issue deleted successfully"}
|
|
except Exception as e:
|
|
await db.rollback()
|
|
raise HTTPException(status_code=500, detail=f"Could not delete issue: {str(e)}")
|