82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi import HTTPException
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.entities.user import UserEntity
|
|
from app.core.usecases.auth_service import create_access_token, verify_token
|
|
from app.core.usecases.user_service import UsersService
|
|
from app.infrastructure.database.dependencies import get_db
|
|
from app.infrastructure.database.models.users import User
|
|
from app.infrastructure.database.repository.user_repository import UsersRepository
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/users", response_model=List[UserEntity])
|
|
def read_users(db: Session = Depends(get_db),
|
|
current_user: User = Depends(verify_token)):
|
|
service = UsersService(db)
|
|
return service.get_all_users()
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=UserEntity)
|
|
def read_user(user_id: int, db: Session = Depends(get_db),
|
|
current_user: User = Depends(verify_token)):
|
|
service = UsersService(db)
|
|
user = service.get_user_by_id(user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return user
|
|
|
|
|
|
@router.post("/users", response_model=UserEntity)
|
|
def create_user(user: UserEntity, db: Session = Depends(get_db),
|
|
current_user: User = Depends(verify_token)):
|
|
service = UsersService(db)
|
|
return service.create_user(user)
|
|
|
|
|
|
@router.put("/users/{user_id}", response_model=UserEntity)
|
|
def update_user(user_id: int, user: UserEntity, db: Session = Depends(get_db),
|
|
current_user: User = Depends(verify_token)):
|
|
service = UsersService(db)
|
|
updated_user = service.update_user(user_id, user)
|
|
if updated_user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return updated_user
|
|
|
|
|
|
@router.delete("/users/{user_id}", response_model=bool)
|
|
def delete_user(user_id: int, db: Session = Depends(get_db),
|
|
current_user: User = Depends(verify_token)):
|
|
service = UsersService(db)
|
|
success = service.delete_user(user_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return success
|
|
|
|
|
|
@router.post("/token")
|
|
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
|
|
user_repo = UsersRepository(db)
|
|
user = user_repo.get_by_login(form_data.username)
|
|
|
|
if not user or user.password != form_data.password:
|
|
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
|
|
|
access_token = create_access_token(data={"sub": user.id})
|
|
|
|
return {
|
|
"access_token": access_token,
|
|
"token_type": "bearer",
|
|
"user": {
|
|
"id": user.id,
|
|
"login": user.login,
|
|
"role_id": user.role.id,
|
|
"role_name": user.role.name,
|
|
}
|
|
}
|