158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
from datetime import datetime
|
|
from typing import List, Optional
|
|
#
|
|
from sqlalchemy.orm import Session
|
|
import logging
|
|
from app.modules.users.models import Users, TipoUsuario, Rol
|
|
#
|
|
from app.modules.users.schema import UserCreate, UserResponse, UpdateUser
|
|
from app.modules.users.repository import UserRepository
|
|
from app.core.security import get_password_hash
|
|
#
|
|
from app.core.auth import get_current_user
|
|
from app.core.baseRepository import BaseRepository
|
|
from app.pipelines.crud import CreatePipe
|
|
|
|
class UserCrudService:
|
|
|
|
@staticmethod
|
|
def create_user(db:Session, user_create: UserCreate, current_user: Optional[Users] = None) -> UserResponse:
|
|
|
|
""" Create a new user in the database and approved if is root user"""
|
|
first = db.query(Users).count()
|
|
is_first_user = (first == 0)
|
|
|
|
user_data = user_create.dict()
|
|
#
|
|
#
|
|
model_data = {
|
|
"name": user_data.get("name"),
|
|
"middle_Name": user_data.get("middle_Name"),
|
|
"last_Name": user_data.get("last_Name"),
|
|
"rfc": user_data.get("rfc"),
|
|
"email": user_data.get("email"),
|
|
"rol_operativo": user_data.get("rol_operativo"),
|
|
"tipo_usuario": user_data.get("tipo_usuario"),
|
|
"permite_diot": user_data.get("permite_diot", False),
|
|
"enterprise_id": user_data.get("enterprise_id"),
|
|
"firma": user_data.get("firma"),
|
|
"is_active": True,
|
|
}
|
|
|
|
#Password comes raw since schema
|
|
plain_pass = user_data.get("password")
|
|
|
|
|
|
# this step is for hash the pasword
|
|
hashed_pass = get_password_hash(plain_pass)
|
|
model_data["password"] = hashed_pass
|
|
|
|
# first user or not that, this is a joke :)
|
|
|
|
|
|
if is_first_user:
|
|
# if first user is being created, it must be root and approved by default
|
|
model_data["tipo_usuario"] = TipoUsuario.ROOT.value
|
|
model_data["created_by"] = None
|
|
|
|
#this point is for make one instance
|
|
new_user = Users(**model_data)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
|
|
else:
|
|
# Case B: aditional User
|
|
if not current_user:
|
|
raise ValueError("requeries authentication for create user")
|
|
|
|
#validate who is creating or if intent create a root user
|
|
if model_data["tipo_usuario"] == TipoUsuario.ROOT.value:
|
|
if current_user.tipo_usuario != TipoUsuario.ROOT.value:
|
|
raise ValueError(" only ROOT user can create another Root user")
|
|
|
|
model_data["created_by"] = current_user.id
|
|
new_user = Users(**model_data)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
|
|
#to this point validate the next users if can be used the app, admin, license_admin and users
|
|
|
|
|
|
|
|
|
|
|
|
return UserResponse.model_validate(new_user)
|
|
# if first is registered allthem users only can set admin or user role.
|
|
|
|
|
|
@staticmethod
|
|
def get_users(db: Session, skip: int =0, limit: int = 100) -> List[UserResponse]:
|
|
""""Get a list of users whith pagination"""
|
|
userRes = db.query(Users).offset(skip).limit(limit).all()
|
|
|
|
return [UserResponse.model_validate(user) for user in userRes]
|
|
|
|
@staticmethod
|
|
def delete_user(db: Session, user_id: int, current_user: Users) -> None:
|
|
""" this is soft delete, for traceability"""
|
|
user = db.query(Users).filter(Users.id == user_id).first()
|
|
|
|
if not user:
|
|
raise ValueError(
|
|
"User not found")
|
|
if user.tipo_usuario == TipoUsuario.ROOT.value:
|
|
raise ValueError("Root user cannot be deleted")
|
|
user.is_active = False
|
|
user.deleted_by = current_user.id
|
|
user.deleted_at = datetime.utcnow()
|
|
|
|
db.commit()
|
|
return {
|
|
"message": f"User has been deleted"
|
|
}
|
|
|
|
#=========== UPDATE =====================
|
|
@staticmethod
|
|
async def Update_user(db: Session, user_id: int, user_data: UpdateUser, current_user: Users) -> Users:
|
|
"""Function to updates users, only personal data"""
|
|
user = db.query(Users).filter(Users.id == user_id).first()
|
|
|
|
if not user:
|
|
raise ValueError(
|
|
"User not found")
|
|
|
|
if current_user.id != user_id:
|
|
|
|
if current_user.tipo_usuario not in ["root", "admin"]:
|
|
raise PermissionError("You don't have permition to updated")
|
|
|
|
if current_user.tipo_usuario == "admin":
|
|
if user.enterprise_id != current_user.enterprise_id:
|
|
raise PermissionError(" Admin only can't update user enterprise ")
|
|
|
|
update_data = user_data.dict(exclude_unset=True)
|
|
|
|
if user.tipo_usuario == "root" and update_data["tipo_usuario"] != "root":
|
|
root_count = db.query(Users).filter(
|
|
Users.tipo_usuario == "root",
|
|
Users.deleted_at.is_(None)
|
|
).count
|
|
if root_count <= 1:
|
|
raise ValueError("Can't downgrade ROOT user")
|
|
|
|
for field, value in update_data.items():
|
|
if hasattr(user, field):
|
|
setattr(user, field, value)
|
|
|
|
user.updated_by = current_user.id
|
|
user.updated_at = datetime.utcnow()
|
|
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
return user
|
|
|
|
|
|
|