agenda/backend-collector
1
0
Fork 0
Code Issues Pull Requests Projects Releases 37 Wiki Activity

search user done

Browse Source
This commit is contained in:
Valentin CZERYBA 2023-10-30 23:37:16 +01:00
parent 4059411828
commit af4af31e26
1 changed files with 33 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
app/routers/users.py
View File

@ -3,6 +3,7 @@ from fastapi.responses import JSONResponse
from datetime import datetime from datetime import datetime
from ..dependencies import users_token, permissions_checker, database from ..dependencies import users_token, permissions_checker, database
from ..models import users from ..models import users
from pydantic import EmailStr
from typing import Annotated from typing import Annotated
from bson import ObjectId from bson import ObjectId
router = APIRouter() router = APIRouter()
@ -10,7 +11,7 @@ router = APIRouter()
@router.get("/users", tags=["users"], response_model=list[users.UserOut]) @router.get("/users", tags=["users"], response_model=list[users.UserOut])
async def read_users(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], skip: int = 0, limit: int = 20): async def read_users(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], skip: int = 0, limit: int = 20, id_user: str | None = None, roles: str | None = None, status: int | None = None, email: EmailStr | None = None):
if limit < 1 or skip < 0 or limit < skip: if limit < 1 or skip < 0 or limit < skip:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,
@ -19,13 +20,42 @@ async def read_users(authorize: Annotated[bool, Depends(permissions_checker.Perm
limit = limit + skip limit = limit + skip
listUsers = [] listUsers = []
user_repository = users.UserRepository(database=database.database) user_repository = users.UserRepository(database=database.database)
for user_index in user_repository.find_by({}, limit=limit, skip=skip): object_search = {}
if status is not None and roles is not None:
object_search = {"$and":[{"roles":{"$eq": roles}}, {"status":{"$eq":status}}]}
else:
if status is not None:
object_search = {"status":{"$eq": status}}
if roles is not None:
object_search = {"roles":{"$eq":roles}}
if id_user is not None:
userid = ObjectId(id_user)
object_search = {"id": {"$regex": userid}}
if status is not None and roles is not None:
object_search = {"$and":[{"id":{"$regex": userid}}, {"roles":{"$eq": roles}}, {"status":{"$eq":status}}]}
else:
if status is not None:
object_search = {"$and":[{"id":{"$regex": userid}}, {"status":{"$eq":status}}]}
if roles is not None:
object_search = {"$and":[{"id":{"$regex": userid}}, {"roles":{"$eq":roles}}]}
if email is not None:
object_search = {"email": {"$eq": email}}
if status is not None and roles is not None:
object_search = {"$and":[{"email":{"$eq": email}}, {"roles":{"$eq": roles}}, {"status":{"$eq":status}}]}
else:
if status is not None:
object_search = {"$and":[{"email":{"$eq": email}}, {"status":{"$eq":status}}]}
if roles is not None:
object_search = {"$and":[{"email":{"$eq": email}}, {"roles":{"$eq":roles}}]}
for user_index in user_repository.find_by(object_search, limit=limit, skip=skip):
user = users.UserOut(id=user_index.id, username=user_index.username, email=user_index.email, status=user_index.status, roles=user_index.roles, firstName=user_index.firstName, name=user_index.name) user = users.UserOut(id=user_index.id, username=user_index.username, email=user_index.email, status=user_index.status, roles=user_index.roles, firstName=user_index.firstName, name=user_index.name)
listUsers.append(user) listUsers.append(user)
return listUsers return listUsers
@router.get("/users/search", tags=["users"], response_model=list[users.UserOut]) @router.get("/users/search", tags=["users"], response_model=list[users.UserOut])
async def read_users_search(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], skip: int = 0, limit: int = 20, key: str | None = None, value: str | None= None): async def read_users_search(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], skip: int = 0, limit: int = 20):
if limit < 1 or skip < 0 or limit < skip: if limit < 1 or skip < 0 or limit < skip:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,