152 lines
5.4 KiB
Python
152 lines
5.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
||
|
||
from typing import Optional, List
|
||
from sqlalchemy.ext.asyncio import AsyncConnection
|
||
|
||
from api.db.connection.session import get_connection_dep
|
||
from api.db.logic.account import (
|
||
create_user,
|
||
get_user_account_page_DTO,
|
||
get_user_by_id,
|
||
get_user_by_login,
|
||
update_user_by_id,
|
||
)
|
||
from api.db.logic.keyring import create_password_key, update_password_key
|
||
from api.db.tables.account import AccountStatus
|
||
from api.schemas.account.account import User
|
||
from api.schemas.base import bearer_schema
|
||
from api.schemas.endpoints.account import AllUserResponse, UserCreate, UserUpdate, UserFilterDTO
|
||
from api.services.auth import get_current_user
|
||
from api.services.user_role_validation import db_user_role_validation
|
||
|
||
api_router = APIRouter(
|
||
prefix="/account",
|
||
tags=["User accountModel"],
|
||
)
|
||
|
||
|
||
@api_router.get("", dependencies=[Depends(bearer_schema)], response_model=AllUserResponse)
|
||
async def get_all_account(
|
||
page: int = Query(1, description="Page number", gt=0),
|
||
limit: int = Query(10, description="КNumber of items per page", gt=0),
|
||
search: Optional[str] = Query(None, description="Search term to filter by name or login or email"),
|
||
status_filter: Optional[List[str]] = Query(None, description="Filter by status"),
|
||
role_filter: Optional[List[str]] = Query(None, description="Filter by role"),
|
||
creator_id: Optional[int] = Query(None, description="Filter by creator id"),
|
||
order_field: Optional[str] = Query("id", description="Field to sort by"),
|
||
order_direction: Optional[str] = Query("asc", description="Sort direction (asc/desc)"),
|
||
connection: AsyncConnection = Depends(get_connection_dep),
|
||
current_user=Depends(get_current_user),
|
||
):
|
||
authorize_user = await db_user_role_validation(connection, current_user)
|
||
|
||
filters = {
|
||
**({"status": status_filter} if status_filter else {}),
|
||
**({"role": role_filter} if role_filter else {}),
|
||
**({"creator_id": [str(creator_id)]} if creator_id else {}),
|
||
}
|
||
|
||
filter_dto = UserFilterDTO(
|
||
pagination={"page": page, "limit": limit},
|
||
search=search,
|
||
order={"field": order_field, "direction": order_direction},
|
||
filters=filters if filters else None,
|
||
)
|
||
|
||
user_list = await get_user_account_page_DTO(connection, filter_dto)
|
||
|
||
if user_list is None:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Accounts not found")
|
||
|
||
return user_list
|
||
|
||
|
||
@api_router.get("/{user_id}", dependencies=[Depends(bearer_schema)], response_model=User)
|
||
async def get_account(
|
||
user_id: int,
|
||
connection: AsyncConnection = Depends(get_connection_dep),
|
||
current_user=Depends(get_current_user),
|
||
):
|
||
authorize_user = await db_user_role_validation(connection, current_user)
|
||
|
||
user = await get_user_by_id(connection, user_id)
|
||
|
||
if user is None:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
|
||
|
||
return user
|
||
|
||
|
||
@api_router.post("", dependencies=[Depends(bearer_schema)], response_model=User)
|
||
async def create_account(
|
||
user: UserCreate,
|
||
connection: AsyncConnection = Depends(get_connection_dep),
|
||
current_user=Depends(get_current_user),
|
||
):
|
||
authorize_user = await db_user_role_validation(connection, current_user)
|
||
|
||
user_validation = await get_user_by_login(connection, user.login)
|
||
|
||
if user_validation is None:
|
||
new_user = await create_user(connection, user, authorize_user.id)
|
||
await create_password_key(connection, user.password, new_user.id)
|
||
return new_user
|
||
else:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST, detail="An account with this information already exists."
|
||
)
|
||
|
||
|
||
@api_router.put("/{user_id}", dependencies=[Depends(bearer_schema)], response_model=UserUpdate)
|
||
async def update_account(
|
||
user_id: int,
|
||
user_update: UserUpdate,
|
||
connection: AsyncConnection = Depends(get_connection_dep),
|
||
current_user=Depends(get_current_user),
|
||
):
|
||
authorize_user = await db_user_role_validation(connection, current_user)
|
||
|
||
user = await get_user_by_id(connection, user_id)
|
||
if user is None:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
|
||
|
||
if user_update.password is not None:
|
||
await update_password_key(connection, user.id, user_update.password)
|
||
|
||
updated_values = user_update.model_dump(by_alias=True, exclude_none=True)
|
||
|
||
if not updated_values:
|
||
return user
|
||
|
||
await update_user_by_id(connection, updated_values, user)
|
||
|
||
user = await get_user_by_id(connection, user_id)
|
||
|
||
return user
|
||
|
||
|
||
@api_router.delete("/{user_id}", dependencies=[Depends(bearer_schema)], response_model=User)
|
||
async def delete_account(
|
||
user_id: int,
|
||
connection: AsyncConnection = Depends(get_connection_dep),
|
||
current_user=Depends(get_current_user),
|
||
):
|
||
authorize_user = await db_user_role_validation(connection, current_user)
|
||
|
||
user = await get_user_by_id(connection, user_id)
|
||
if user is None:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
|
||
|
||
user_update = UserUpdate(status=AccountStatus.DELETED.value)
|
||
|
||
updated_values = user_update.model_dump(by_alias=True, exclude_none=True)
|
||
|
||
if not updated_values:
|
||
return user
|
||
|
||
await update_user_by_id(connection, updated_values, user)
|
||
|
||
user = await get_user_by_id(connection, user_id)
|
||
|
||
return user
|