mirror of
https://gitlab.com/MoonTestUse1/AdministrationItDepartmens.git
synced 2025-08-14 00:25:46 +02:00
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
"""Employee endpoints."""
|
||
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from sqlalchemy.orm import Session
|
||
from typing import List
|
||
|
||
from app.database import get_db
|
||
from app.models.user import User
|
||
from app.core.auth import get_current_user, get_password_hash
|
||
from app.schemas.user import User as UserSchema, UserCreate, UserUpdate
|
||
|
||
router = APIRouter()
|
||
|
||
@router.post("/", response_model=UserSchema)
|
||
def create_employee(
|
||
user: UserCreate,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""Create new employee (admin only)."""
|
||
if not current_user.is_admin:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="Not enough permissions"
|
||
)
|
||
|
||
# Проверяем, не существует ли пользователь с таким email
|
||
db_user = db.query(User).filter(User.email == user.email).first()
|
||
if db_user:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Email already registered"
|
||
)
|
||
|
||
# Создаем пользователя
|
||
hashed_password = get_password_hash(user.password)
|
||
db_user = User(
|
||
email=user.email,
|
||
full_name=user.full_name,
|
||
hashed_password=hashed_password,
|
||
is_admin=user.is_admin
|
||
)
|
||
db.add(db_user)
|
||
db.commit()
|
||
db.refresh(db_user)
|
||
return db_user
|
||
|
||
@router.get("/me", response_model=UserSchema)
|
||
def get_current_employee(current_user: User = Depends(get_current_user)):
|
||
"""Get current employee info."""
|
||
return current_user
|
||
|
||
@router.put("/me", response_model=UserSchema)
|
||
def update_employee_me(
|
||
user_update: UserUpdate,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""Update current employee info."""
|
||
update_data = user_update.model_dump(exclude_unset=True)
|
||
if "password" in update_data:
|
||
update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
|
||
|
||
for key, value in update_data.items():
|
||
setattr(current_user, key, value)
|
||
|
||
db.commit()
|
||
db.refresh(current_user)
|
||
return current_user
|
||
|
||
@router.get("/{employee_id}", response_model=UserSchema)
|
||
def get_employee(
|
||
employee_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""Get employee by ID (admin only)."""
|
||
if not current_user.is_admin:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="Not enough permissions"
|
||
)
|
||
|
||
employee = db.query(User).filter(User.id == employee_id).first()
|
||
if not employee:
|
||
raise HTTPException(status_code=404, detail="Employee not found")
|
||
return employee |