| @@ -8,7 +8,14 @@ from typing import List, Annotated | |||
| from apis.v1.route_auth import get_current_user | |||
| from core.config import settings | |||
| from db.models.user import User | |||
| from schemas.user import UserCreate, ShowUser, ShowDriver, DriverCreate, OutputUser, UsersPage | |||
| from schemas.user import ( | |||
| UserCreate, | |||
| ShowUser, | |||
| ShowDriver, | |||
| DriverCreate, | |||
| OutputUser, | |||
| UsersPage, | |||
| ) | |||
| from db.session import get_db | |||
| from db.repository.user import ( | |||
| create_new_user, | |||
| @@ -17,11 +24,11 @@ from db.repository.user import ( | |||
| replace_user_data, | |||
| create_new_driver, | |||
| delete_user_data, | |||
| get_users_by_name | |||
| get_users_by_name, | |||
| user_search_query, | |||
| ) | |||
| router = APIRouter() | |||
| @@ -133,25 +140,11 @@ def delete_user( | |||
| @router.get("/search/", response_model=UsersPage) | |||
| def search_users(db: Session = Depends(get_db), name: str = None, role: str = None, page: int = 1, per_page: int = 20): | |||
| query = db.query(User).filter(User.Name.like(f"{name}%")) | |||
| if role is not None and role != "Admin": | |||
| query = query.filter(User.Role == role) | |||
| total_users = query.count() | |||
| total_pages = ceil(total_users / per_page) | |||
| users = query.offset((page - 1) * per_page).limit(per_page).all() | |||
| output_users = [OutputUser( | |||
| id= user.Id, | |||
| Name=user.Name, | |||
| MiddleName = user.MiddleName, | |||
| LastName=user.LastName, | |||
| ContactNumber=user.ContactNumber, | |||
| Address=user.Address, | |||
| Email=user.Email, | |||
| Role=user.Role, | |||
| AssignedVehicle= None | |||
| ) for user in users] | |||
| return {"users": output_users, "total_pages": total_pages} | |||
| def search_users( | |||
| db: Session = Depends(get_db), | |||
| name: str = None, | |||
| role: str = None, | |||
| page: int = 1, | |||
| per_page: int = 20, | |||
| ): | |||
| return user_search_query(db=db, name=name, role=role, page=page, per_page=per_page) | |||
| @@ -1,7 +1,8 @@ | |||
| # Creating a new user in the database | |||
| from math import ceil | |||
| from sqlalchemy.orm import Session | |||
| from schemas.user import UserCreate, DriverCreate | |||
| from schemas.user import OutputUser, UserCreate, DriverCreate | |||
| from db.models.user import User | |||
| from core.hashing import Hasher | |||
| from db.models.drivetask import DriveTask | |||
| @@ -83,7 +84,10 @@ def list_users(db: Session, role: str = "Any"): | |||
| users = db.query(User).filter((User.Role == role) | (role == "Any")).all() | |||
| return users | |||
| def get_users_by_name(db: Session, name: str = "", role: str = None, page: int = 1, per_page: int = 20): | |||
| def get_users_by_name( | |||
| db: Session, name: str = "", role: str = None, page: int = 1, per_page: int = 20 | |||
| ): | |||
| if role == "Admin": | |||
| return None | |||
| if role is None: | |||
| @@ -93,6 +97,7 @@ def get_users_by_name(db: Session, name: str = "", role: str = None, page: int = | |||
| users = users.offset((page - 1) * per_page).limit(per_page).all() | |||
| return users | |||
| def replace_user_data(user_id: int, user_data: UserCreate, db: Session): | |||
| user = db.query(User).filter(User.Id == user_id).first() | |||
| if not user: | |||
| @@ -121,3 +126,31 @@ def delete_user_data(id: int, db: Session): | |||
| db.delete(user) | |||
| db.commit() | |||
| return "userDeleted" | |||
| def user_search_query( | |||
| db: Session, name: str = "", role: str = None, page: int = 1, per_page: int = 20 | |||
| ): | |||
| query = db.query(User).filter(User.Name.like(f"{name}%")) | |||
| if role is not None and role != "Admin": | |||
| query = query.filter(User.Role == role) | |||
| total_users = query.count() | |||
| total_pages = ceil(total_users / per_page) | |||
| users = query.offset((page - 1) * per_page).limit(per_page).all() | |||
| output_users = [ | |||
| OutputUser( | |||
| id=user.Id, | |||
| Name=user.Name, | |||
| MiddleName=user.MiddleName, | |||
| LastName=user.LastName, | |||
| ContactNumber=user.ContactNumber, | |||
| Address=user.Address, | |||
| Email=user.Email, | |||
| Role=user.Role, | |||
| AssignedVehicle=None, | |||
| ) | |||
| for user in users | |||
| ] | |||
| return {"users": output_users, "total_pages": total_pages} | |||
| @@ -46,15 +46,19 @@ class OutputUser(BaseModel): | |||
| Address: str | |||
| Email: EmailStr | |||
| Role: str | |||
| AssignedVehicle: Optional[str] # replace str with the actual type of AssignedVehicle | |||
| AssignedVehicle: Optional[ | |||
| str | |||
| ] # replace str with the actual type of AssignedVehicle | |||
| class Config: | |||
| orm_mode = True | |||
| class UsersPage(BaseModel): | |||
| users: List[OutputUser] | |||
| total_pages: int | |||
| class ShowDriverNoVehicle(ShowUser): | |||
| DrivingLicenseNumber: str | None | |||