dip test

backend/app/crud/employees.py

@@ -1,9 +1,13 @@
-"""Employee CRUD operations2"""
-from typing import Optional, Dict, Any
+"""Employee CRUD operations"""
 from sqlalchemy.orm import Session
-
+from typing import List, Optional
 from ..models.employee import Employee
-from ..utils.security import get_password_hash
+from ..schemas.employee import EmployeeCreate, EmployeeUpdate
+from ..utils.loggers import auth_logger
+
+def get_employees(db: Session, skip: int = 0, limit: int = 100) -> List[Employee]:
+    """Get all employees"""
+    return db.query(Employee).offset(skip).limit(limit).all()
 
 def get_employee(db: Session, employee_id: int) -> Optional[Employee]:
     """Get employee by ID"""
@@ -16,49 +20,35 @@ def get_employee_by_credentials(db: Session, first_name: str, last_name: str) ->
         Employee.last_name == last_name
     ).first()
 
-def get_employee_by_login(db: Session, login: str) -> Optional[Employee]:
-    """Get employee by login"""
-    return db.query(Employee).filter(Employee.login == login).first()
-
-def get_employees(db: Session, skip: int = 0, limit: int = 100):
-    """Get list of employees"""
-    return db.query(Employee).offset(skip).limit(limit).all()
-
-def create_employee(db: Session, employee_data: Dict[str, Any]) -> Employee:
+def create_employee(db: Session, employee: EmployeeCreate, hashed_password: str) -> Employee:
     """Create new employee"""
-    # Хешируем пароль
-    hashed_password = get_password_hash(employee_data["password"])
-    
-    # Создаем сотрудника
     db_employee = Employee(
-        login=employee_data.get("login"),
-        first_name=employee_data["first_name"],
-        last_name=employee_data["last_name"],
-        department=employee_data["department"],
-        office=employee_data["office"],
+        first_name=employee.first_name,
+        last_name=employee.last_name,
+        department=employee.department,
+        office=employee.office,
         hashed_password=hashed_password,
-        is_admin=employee_data.get("is_admin", False)
+        is_admin=employee.is_admin
     )
     db.add(db_employee)
     db.commit()
     db.refresh(db_employee)
     return db_employee
 
-def update_employee(db: Session, employee: Employee, employee_data: Dict[str, Any]) -> Employee:
-    """Update employee"""
-    # Если есть пароль в данных, хешируем его
-    if "password" in employee_data:
-        employee_data["hashed_password"] = get_password_hash(employee_data.pop("password"))
-    
-    # Обновляем поля
-    for field, value in employee_data.items():
-        setattr(employee, field, value)
-    
-    db.commit()
-    db.refresh(employee)
-    return employee
+def update_employee(db: Session, employee_id: int, employee: EmployeeUpdate) -> Optional[Employee]:
+    """Update employee data"""
+    db_employee = get_employee(db, employee_id)
+    if db_employee:
+        for key, value in employee.dict(exclude_unset=True).items():
+            setattr(db_employee, key, value)
+        db.commit()
+        db.refresh(db_employee)
+    return db_employee
 
-def delete_employee(db: Session, employee: Employee) -> None:
+def delete_employee(db: Session, employee_id: int) -> Optional[Employee]:
     """Delete employee"""
-    db.delete(employee)
-    db.commit()
+    db_employee = get_employee(db, employee_id)
+    if db_employee:
+        db.delete(db_employee)
+        db.commit()
+    return db_employee