Files

63 lines
2.2 KiB
Python

from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.credits.models import Credits
from app.modules.credits.schema import CreditsCreate, CreditsUpdate, CerditsResponse, MessageResponse
class CreditsService:
@staticmethod
def create_credits(db: Session, data:CreditsCreate ):
new_credits = Credits(**data.dict())
db.add(new_credits)
db.commit()
db.refresh(new_credits)
return new_credits
@staticmethod
def get_credits(db: Session, credits_id: int, current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
return credits
@staticmethod
def update_credits(db: Session, credits_id: int, data:CreditsUpdate , current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este credits")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(credits, key, value)
db.commit()
db.refresh(credits)
return credits
@staticmethod
def delete_credits(db: Session, credits_id: int, current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este credits")
try:
credits.is_active = False
credits.deleted_at = datetime.utcnow()
credits.deleted_by = current_user.id
db.commit()
db.refresh(credits)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el credits: {e}")
return {"message": "credits eliminado correctamente"}