Multimedika/Bot_Development

1from api.router.user import user_dependency2from fastapi.responses import JSONResponse3from fastapi import APIRouter, Depends4 5from db.models import Category6from db.database import get_db7from db.query.query_category import CategoryQuery8from service.dto import CategoryCreate9 10from script.vector_db import IndexManager11from sqlalchemy.orm import Session12from sqlalchemy.exc import SQLAlchemyError, IntegrityError13 14from typing import Annotated15 16router = APIRouter(tags=["Category"])17 18index_manager = IndexManager()19db_dependency = Annotated[Session, Depends(get_db)]20 21 22@router.get("/category")23async def get_all_categories_router(user: user_dependency, db: db_dependency):24    if user is None or user.get("role_id") != 1:25        return JSONResponse(status_code=401, content="Authentication Failed")26 27    try:28        # Logic to retrieve all categories29        category_query = CategoryQuery(user)30        categories = category_query.get_all_categories(db)31        32        print(categories)33        34        35        # categories = db.query(Category).all()36        if not categories:37            return JSONResponse(status_code=404, content="No categories found")38 39        return {40            "message": "Categories retrieved successfully",41            "categories": [42                {"id": cat.id, "category": cat.category} for cat in categories43            ],44        }45 46    except SQLAlchemyError as e:47        return JSONResponse(48            status_code=500, content="Database error occurred: " + str(e)49        )50 51 52@router.get("/category/{category_id}")53async def get_category_by_id_router(54    user: user_dependency,55    db: db_dependency,56    category_id: int,57):58    if user is None or user.get("role_id") != 1:59        return JSONResponse(status_code=401, content="Authentication Failed")60 61    try:62        # Fetch categories based on the list of provided category_ids63        category_query = CategoryQuery(user)64        category = category_query.get_category_by_id(db, category_id)65 66        if category is None:67            return JSONResponse(status_code=404, content="No categories found for the given IDs")68 69        return {70            "message": "Categories retrieved successfully",71            "category": {"id": category.id, "category": category.category},72        }73 74    except SQLAlchemyError as e:75        return JSONResponse(76            status_code=500, content="Database error occurred: " + str(e)77        )78 79@router.post("/category")80async def create_category(user: user_dependency, db: db_dependency, category: CategoryCreate):81    if user is None or user.get("role_id") != 1:82        return JSONResponse(status_code=401, content="Authentication Failed")83 84    try:85        # Check if category already exists86        category_query = CategoryQuery(user)87        existing_category = category_query.get_existing_category(db, category.category_name)88        print(existing_category)89        if existing_category:  # Check if the category already exists90            return JSONResponse(status_code=400, content={"error": "Category already exists"})91        92        # Add category93        category_query.add_category(db, category.category_name)94        print("category added")95 96        return {97            "message": "Category created successfully",98        }99 100    except IntegrityError:101        db.rollback()102        return JSONResponse(103            status_code=400,104            content="Database integrity error: possibly a duplicate entry.",105        )106 107@router.put("/category/{category_id}")108async def update_category(109    user: user_dependency, db: db_dependency, category_id: int, category: CategoryCreate110):111    if user is None or user.get("role_id") != 1:112        return JSONResponse(status_code=401, content="Authentication Failed")113 114    try:115        # Logic to update an existing category116        existing_category = (117            db.query(Category).filter(Category.id == category_id).first()118        )119        if not existing_category:120            return JSONResponse(status_code=404, content="Category not found")121 122        existing_category.category = category.category_name123        db.commit()124 125        return {"message": "Category updated successfully"}126 127    except SQLAlchemyError as e:128        db.rollback()129        return JSONResponse(130            status_code=500, content="Database error occurred: " + str(e)131        )132 133 134@router.delete("/category/{category_id}")135async def delete_category(user: user_dependency, db: db_dependency, category_id: int):136    if user is None or user.get("role_id") != 1:137        return JSONResponse(status_code=401, content="Authentication Failed")138 139    try:140        # Logic to delete an existing category141        existing_category = (142            db.query(Category).filter(Category.id == category_id).first()143        )144        if not existing_category:145            return JSONResponse(status_code=404, content="Category not found")146 147        db.delete(existing_category)148        db.commit()149 150        return {"message": "Category deleted successfully"}151 152    except SQLAlchemyError as e:153        db.rollback()154        return JSONResponse(155            status_code=500, content="Database error occurred: " + str(e)156        )