| @@ -1,9 +1,19 @@ | |||
| # Base API router -- collecting all APIs here to not clutter main.py | |||
| from fastapi import APIRouter | |||
| from apis.v1 import route_user, route_vehicle, route_auth | |||
| from apis.v1 import ( | |||
| route_user, | |||
| route_vehicle, | |||
| route_auth, | |||
| route_task, | |||
| route_maintenancejob, | |||
| ) | |||
| api_router = APIRouter() | |||
| api_router.include_router(route_user.router, prefix="/user", tags=["users"]) | |||
| api_router.include_router(route_vehicle.router, prefix="/vehicle", tags=["vehicles"]) | |||
| api_router.include_router(route_auth.router, prefix="", tags=["auth"]) | |||
| api_router.include_router(route_task.router, prefix="/task", tags=["task"]) | |||
| api_router.include_router( | |||
| route_maintenancejob.router, prefix="/maintenancejob", tags=["maintenancejob"] | |||
| ) | |||
| @@ -0,0 +1,95 @@ | |||
| from fastapi import Depends, APIRouter, HTTPException, status | |||
| from sqlalchemy.orm import Session | |||
| from db.session import get_db | |||
| from db.repository.maintenancejob import ( | |||
| create_new_maintenancejob, | |||
| create_car_part, | |||
| get_all_maintenance_jobs, | |||
| get_maintenance_job, | |||
| change_maintenance_status | |||
| ) | |||
| from schemas.maintenancejob import CreateMaintenanceJob | |||
| from schemas.carpart import CreateCarPart | |||
| from db.models.user import User | |||
| from apis.v1.route_auth import get_current_user | |||
| router = APIRouter() | |||
| @router.post("/", status_code=status.HTTP_201_CREATED) | |||
| def create_maintenancejob( | |||
| maintenancejob: CreateMaintenanceJob, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Maintenance" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| maintenancejob_res = create_new_maintenancejob( | |||
| maintenancejob=maintenancejob, maintenanceworker=current_user.Id, db=db | |||
| ) | |||
| return maintenancejob_res | |||
| @router.post("/carpart", status_code=status.HTTP_201_CREATED) | |||
| def create_carpart( | |||
| car_part: CreateCarPart, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Maintenance" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| car_part_res = create_car_part(car_part=car_part, db=db) | |||
| return car_part_res | |||
| @router.get("/", status_code=status.HTTP_200_OK) | |||
| def get_all_maintenancejobs( | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Maintenance" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| maintenancejobs = get_all_maintenance_jobs(db) | |||
| return maintenancejobs | |||
| @router.get("/{maintenance_job_id}", status_code=status.HTTP_200_OK) | |||
| def get_maintenancejob( | |||
| maintenance_job_id: int, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Maintenance" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| maintenancejob = get_maintenance_job(maintenance_job_id, db) | |||
| return maintenancejob | |||
| @router.patch("/{maintenance_job_id}", status_code=status.HTTP_200_OK) | |||
| def change_status( | |||
| maintenance_job_id: int, | |||
| status: str, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Maintenance" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| result = change_maintenance_status(maintenance_job_id, status, db) | |||
| if result is None: | |||
| raise HTTPException( | |||
| status_code=404, detail="Maintenance job not found" | |||
| ) | |||
| return result | |||
| @@ -0,0 +1,120 @@ | |||
| from fastapi import Depends, APIRouter, HTTPException, status | |||
| from sqlalchemy.orm import Session | |||
| from db.session import get_db | |||
| from core.config import settings | |||
| from db.repository.drivetask import ( | |||
| create_new_task, | |||
| get_task_driver, | |||
| change_task_status, | |||
| get_all_tasks, | |||
| get_task_by_id, | |||
| get_tasks_by_driver, | |||
| ) | |||
| from schemas.drivetask import CreateTask | |||
| from db.models.user import User | |||
| from apis.v1.route_auth import get_current_user | |||
| router = APIRouter() | |||
| @router.post("/", status_code=status.HTTP_201_CREATED) | |||
| def create_task( | |||
| task: CreateTask, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| task_res = create_new_task(task=task, db=db) | |||
| if task_res == "notdriver": | |||
| raise HTTPException( | |||
| status_code=404, detail=f"Driver with id {task.DriverId} not found" | |||
| ) | |||
| return task | |||
| @router.patch("/", status_code=status.HTTP_200_OK) | |||
| def changeStatus( | |||
| task_id: int, | |||
| status: str, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role == "Admin" or current_user.Role == "Driver": | |||
| if status not in settings.ALLOWED_TASK_STATUS: | |||
| raise HTTPException( | |||
| status_code=400, | |||
| detail=f"Status {status} is not allowed. Allowed status are {settings.ALLOWED_TASK_STATUS}", | |||
| ) | |||
| if current_user.Role == "Driver": | |||
| verification = get_task_driver(task_id, db) | |||
| if verification.Id != current_user.Id: | |||
| raise HTTPException( | |||
| status_code=403, | |||
| detail="You are not authorized to perform this action", | |||
| ) | |||
| task = change_task_status(task_id, status, db) | |||
| if task == "notaskfound": | |||
| raise HTTPException( | |||
| status_code=404, detail=f"Task with id {task_id} not found" | |||
| ) | |||
| return task | |||
| else: | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| @router.get("/", status_code=status.HTTP_200_OK) | |||
| def getAllTasks( | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role == "Admin": | |||
| tasks = get_all_tasks(db) | |||
| return tasks | |||
| else: | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| @router.get("/{task_id}", status_code=status.HTTP_200_OK) | |||
| def getTaskById( | |||
| task_id: int, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| task = get_task_by_id(task_id, db) | |||
| if task == "notaskfound": | |||
| raise HTTPException(status_code=404, detail=f"Task with id {task_id} not found") | |||
| return task | |||
| @router.get("/driver/{driver_id}", status_code=status.HTTP_200_OK) | |||
| def getTasksByDriver( | |||
| driver_id: int, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Admin" and current_user.Role != "Driver": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| if current_user.Role == "Driver": | |||
| if current_user.Id != driver_id: | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| tasks = get_tasks_by_driver(driver_id, db) | |||
| if tasks == "notdriver": | |||
| raise HTTPException( | |||
| status_code=404, detail=f"Driver with id {driver_id} not found" | |||
| ) | |||
| return tasks | |||
| @@ -7,7 +7,7 @@ from apis.v1.route_auth import get_current_user | |||
| from db.models.user import User | |||
| from schemas.user import UserCreate, ShowUser | |||
| from db.session import get_db | |||
| from db.repository.user import create_new_user, list_users, get_user_by_id | |||
| from db.repository.user import create_new_user, list_users, get_user_by_id, replace_user_data | |||
| router = APIRouter() | |||
| @@ -35,6 +35,18 @@ def get_all_users(db: Session = Depends(get_db), role: str = None): | |||
| users = list_users(db=db, role=role) | |||
| return users | |||
| @router.put("/{user_id}", response_model=ShowUser, status_code=status.HTTP_202_ACCEPTED) | |||
| def update_user( | |||
| user_id: int, | |||
| user: UserCreate, | |||
| db: Session = Depends(get_db), | |||
| current_user: User = Depends(get_current_user), | |||
| ): | |||
| if current_user.Role != "Admin": | |||
| raise HTTPException(status_code=403, detail="You are not authorized to perform this action") | |||
| user = replace_user_data(user_id=user_id, user=user, db=db) | |||
| return user | |||
| @router.get("/me", response_model=ShowUser, status_code=status.HTTP_200_OK) | |||
| def get_user_me( | |||
| @@ -90,6 +90,8 @@ async def assign_driver( | |||
| @router.get("/", response_model=List[OutputVehicle], status_code=status.HTTP_200_OK) | |||
| async def get_all_vehicles(db: Session = Depends(get_db)): | |||
| vehicles = list_vehicles(db=db) | |||
| if vehicles == []: | |||
| raise HTTPException(status_code=404, detail="No vehicles found") | |||
| return vehicles | |||
| @@ -151,11 +153,11 @@ def update_vehicle_location( | |||
| ): | |||
| print(current_user) | |||
| print(current_user.Name) | |||
| if current_user.Role != "Driver": | |||
| if current_user.Role != "Driver" and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not authorized to perform this action" | |||
| ) | |||
| if current_user.AssignedVehicle != vehicle_id: | |||
| if current_user.AssignedVehicle != vehicle_id and current_user.Role != "Admin": | |||
| raise HTTPException( | |||
| status_code=403, detail="You are not the correct car driver" | |||
| ) | |||
| @@ -10,6 +10,7 @@ class Settings: | |||
| ACCESS_TOKEN_EXPIRE: int = 60 * 24 * 7 # 7 days | |||
| SECRET_KEY: str = "tH357aC6oA7ofCaN3yTffYkRh" | |||
| ALGORITHM: str = "HS256" | |||
| ALLOWED_TASK_STATUS: list = ["Pending", "In Progress", "Completed", "Cancelled"] | |||
| settings = Settings() | |||
| @@ -29,7 +30,8 @@ def createAdminAcc(): | |||
| LastName="Turgunov", | |||
| ContactNumber="+77071234567", | |||
| Role="Admin", | |||
| BirthDate="2000-01-01T00:00:00+06:00", | |||
| GovernmentId="123456789", | |||
| Address="Nazarbayev University", | |||
| ) | |||
| if db.query(User).filter(User.Email == user.Email).first(): | |||
| return False | |||
| @@ -2,3 +2,6 @@ | |||
| from db.base_class import Base | |||
| from db.models.user import User | |||
| from db.models.vehicle import Vehicle | |||
| from db.models.carpart import CarPart | |||
| from db.models.maintenancejob import MaintenanceJob | |||
| from db.models.drivetask import DriveTask | |||
| @@ -0,0 +1,16 @@ | |||
| from sqlalchemy import Column, Integer, String, ForeignKey, DateTime | |||
| from sqlalchemy.orm import relationship | |||
| from db.base import Base | |||
| class Auction(Base): | |||
| Id = Column(Integer, primary_key=True, index=True) | |||
| initialPrice = Column(Integer, nullable=False) | |||
| minimalBet = Column(Integer, nullable=False) | |||
| carID = Column(Integer, ForeignKey("car.Id"), nullable=False) | |||
| vehicle = relationship("Vehicle", back_populates="auction") | |||
| CreatedBy = relationship("User", back_populates="auctions") | |||
| dateStart = Column(DateTime, nullable=False) | |||
| dateEnd = Column(DateTime, nullable=False) | |||
| @@ -0,0 +1,15 @@ | |||
| from sqlalchemy import Column, Integer, String, ForeignKey | |||
| from sqlalchemy.orm import relationship | |||
| from db.base import Base | |||
| class CarPart(Base): | |||
| Id = Column(Integer, primary_key=True, index=True) | |||
| # a list of weak entities of class CarPart | |||
| ParentId = Column(Integer, ForeignKey("maintenancejob.Id"), nullable=False) | |||
| parent = relationship("MaintenanceJob", back_populates="CarParts") | |||
| Name = Column(String, nullable=False) | |||
| Number = Column(String, nullable=False) | |||
| Condition = Column(String, nullable=False) | |||
| ImageURL = Column(String, nullable=False) | |||
| Cost = Column(Integer, nullable=False) | |||
| @@ -0,0 +1,13 @@ | |||
| from sqlalchemy import Column, Integer, String, ForeignKey, ARRAY | |||
| from db.base import Base | |||
| from sqlalchemy.orm import relationship | |||
| class DriveTask(Base): | |||
| Id = Column(Integer, primary_key=True, index=True) | |||
| DriverId = Column(ForeignKey("user.Id"), nullable=False) | |||
| CreatedBy = relationship("User", back_populates="driveTasks") | |||
| Description = Column(String, nullable=True) | |||
| Status = Column(String, nullable=False) | |||
| StartLocation = Column(ARRAY(String), nullable=False) | |||
| EndLocation = Column(ARRAY(String), nullable=False) | |||
| @@ -0,0 +1,13 @@ | |||
| from sqlalchemy import Column, Integer, String, DateTime, ForeignKey | |||
| from sqlalchemy.orm import relationship | |||
| from db.base import Base | |||
| class MaintenanceJob(Base): | |||
| Id = Column(Integer, primary_key=True, index=True) | |||
| # a list of weak entities of class CarPart | |||
| CarParts = relationship("CarPart", back_populates="parent") | |||
| CreatedBy = relationship("MaintenancePerson", back_populates="maintenanceJobs") | |||
| Description = Column(String, nullable=False) | |||
| Date = Column(DateTime, nullable=False) | |||
| MaintenanceWorker = Column(ForeignKey("user.Id"), nullable=False) | |||
| @@ -1,5 +1,5 @@ | |||
| # PostgreSQL table model for users | |||
| from sqlalchemy import Column, Integer, String, DateTime, ForeignKey | |||
| from sqlalchemy import Column, Integer, String, ForeignKey | |||
| from sqlalchemy.orm import relationship | |||
| from db.base import Base | |||
| @@ -9,11 +9,23 @@ class User(Base): | |||
| Name = Column(String, nullable=False) | |||
| MiddleName = Column(String, nullable=True) | |||
| LastName = Column(String, nullable=False) | |||
| BirthDate = Column(DateTime, nullable=False) | |||
| GovernmentId = Column(String, nullable=False) | |||
| Address = Column(String, nullable=False) | |||
| ContactNumber = Column(String, nullable=False) | |||
| Email = Column(String, nullable=False) | |||
| Role = Column(String, nullable=False) | |||
| HashedPassword = Column(String, nullable=False) | |||
| #Admin-specific relationships | |||
| CreatedAuctions = relationship("Auction", back_populates="CreatedBy") | |||
| #Driver-specific relationships | |||
| DrivingLicenseNumber = Column(String, nullable=True) | |||
| AssignedVehicle = Column(Integer, ForeignKey("vehicle.Id"), nullable=True) | |||
| DriveTasks = relationship("DriveTask", back_populates="CreatedBy") | |||
| vehicle = relationship("Vehicle", back_populates="driver") | |||
| HashedPassword = Column(String, nullable=False) | |||
| #MaintenancePerson-specific relationships | |||
| maintenanceJobs = relationship("MaintenanceJob", back_populates="CreatedBy") | |||
| #FuelingPerson-specific relationships | |||
| fuelingJobs = relationship("FuelingJob", back_populates="CreatedBy") | |||
| @@ -14,10 +14,11 @@ class Vehicle(Base): | |||
| Model = Column(String, nullable=False) | |||
| Year = Column(Integer, nullable=False) | |||
| LicensePlate = Column(String, nullable=False) | |||
| Type = Column(String, nullable=False) | |||
| CurrentLocation = Column(ARRAY(String), nullable=True) | |||
| Fuel = Column(Integer, nullable=False) | |||
| Mileage = Column(Integer, nullable=False) | |||
| Status = Column(String, nullable=False) | |||
| Capacity = Column(Integer, nullable=False) | |||
| MaintenanceNotes = Column(ARRAY(String), nullable=True) | |||
| DriverHistory = Column(ARRAY(Integer), nullable=True) | |||
| driver = relationship("User", back_populates="vehicle") | |||
| @@ -0,0 +1,66 @@ | |||
| from sqlalchemy.orm import Session | |||
| from schemas.drivetask import CreateTask | |||
| from db.models.drivetask import DriveTask | |||
| from db.repository.user import get_user_by_id | |||
| def create_new_task(task: CreateTask, db: Session): | |||
| driver = get_user_by_id(task.DriverId, db) | |||
| if driver is None: | |||
| return "notdriver" | |||
| elif driver.Role != "Driver": | |||
| return "notdriver" | |||
| task_object = DriveTask( | |||
| DriverId=task.DriverId, | |||
| Description=task.Description, | |||
| Status="Pending", | |||
| StartLocation=task.StartLocation, | |||
| EndLocation=task.EndLocation, | |||
| ) | |||
| db.add(task_object) | |||
| db.commit() | |||
| db.refresh(task_object) | |||
| return task_object | |||
| def change_task_status(task_id: int, status: str, db: Session): | |||
| task = db.query(DriveTask).filter(DriveTask.Id == task_id).first() | |||
| if not task: | |||
| return "notaskfound" | |||
| task.Status = status | |||
| db.commit() | |||
| db.refresh(task) | |||
| return task | |||
| def get_task_driver(task_id: int, db: Session): | |||
| task = db.query(DriveTask).filter(DriveTask.Id == task_id).first() | |||
| if not task: | |||
| return "notaskfound" | |||
| driver = get_user_by_id(task.DriverId, db) | |||
| if not driver: | |||
| return "notdriver" | |||
| return driver | |||
| def get_tasks_by_driver(driver_id: int, db: Session): | |||
| driver = get_user_by_id(driver_id, db) | |||
| if not driver: | |||
| return "notdriver" | |||
| if driver.Role != "Driver": | |||
| return "notdriver" | |||
| tasks = db.query(DriveTask).filter(DriveTask.DriverId == driver_id).all() | |||
| return tasks | |||
| def get_task_by_id(task_id: int, db: Session): | |||
| task = db.query(DriveTask).filter(DriveTask.Id == task_id).first() | |||
| if not task: | |||
| return "notaskfound" | |||
| return task | |||
| def get_all_tasks(db: Session): | |||
| tasks = db.query(DriveTask).all() | |||
| return tasks | |||
| @@ -0,0 +1,74 @@ | |||
| from sqlalchemy.orm import Session | |||
| from schemas.maintenancejob import CreateMaintenanceJob | |||
| from db.models.maintenancejob import MaintenanceJob | |||
| from schemas.carpart import CreateCarPart | |||
| from db.models.carpart import CarPart | |||
| def create_new_maintenancejob( | |||
| maintenancejob: CreateMaintenanceJob, maintenanceworker, db: Session | |||
| ): | |||
| maintenancejob_object = MaintenanceJob( | |||
| MaintenanceWorker=maintenanceworker, | |||
| Description=maintenancejob.Description, | |||
| Date=maintenancejob.Date, | |||
| ) | |||
| print("OBJECT CREATED") | |||
| db.add(maintenancejob_object) | |||
| db.commit() | |||
| db.refresh(maintenancejob_object) | |||
| return maintenancejob_object | |||
| def create_car_part(car_part: CreateCarPart, db: Session): | |||
| car_part_object = CarPart( | |||
| ParentId=car_part.ParentId, | |||
| Name=car_part.Name, | |||
| Number=car_part.Number, | |||
| Condition=car_part.Condition, | |||
| ImageURL=car_part.ImageURL, | |||
| Cost=car_part.Cost, | |||
| ) | |||
| print("OBJECT CREATED") | |||
| db.add(car_part_object) | |||
| db.commit() | |||
| db.refresh(car_part_object) | |||
| return car_part_object | |||
| def calculate_total_cost(car_parts: CarPart): | |||
| total_cost = 0 | |||
| for part in car_parts: | |||
| total_cost += part["Cost"] | |||
| return total_cost | |||
| def get_all_maintenance_jobs(db: Session): | |||
| maintenancejobs = db.query(MaintenanceJob).all() | |||
| result = [] | |||
| for job in maintenancejobs: | |||
| job_dict = job.__dict__ | |||
| job_dict["CarParts"] = [part.__dict__ for part in job.CarParts] | |||
| job_dict["TotalCost"] = calculate_total_cost(job.CarParts) | |||
| result.append(job_dict) | |||
| return maintenancejobs | |||
| def get_maintenance_job(maintenancejob_id: int, db: Session): | |||
| maintenancejob = ( | |||
| db.query(MaintenanceJob).filter(MaintenanceJob.Id == maintenancejob_id).first() | |||
| ) | |||
| result = maintenancejob.__dict__ | |||
| result["CarParts"] = [part.__dict__ for part in maintenancejob.CarParts] | |||
| return result | |||
| def change_maintenance_status(maintenancejob_id: int, status: str, db: Session): | |||
| maintenancejob = ( | |||
| db.query(MaintenanceJob).filter(MaintenanceJob.Id == maintenancejob_id).first() | |||
| ) | |||
| if maintenancejob is None: | |||
| return None | |||
| maintenancejob.Status = status | |||
| db.commit() | |||
| db.refresh(maintenancejob) | |||
| return maintenancejob | |||
| @@ -12,7 +12,8 @@ def create_new_user(user: UserCreate, db: Session): | |||
| Name=user.Name, | |||
| MiddleName=user.MiddleName, | |||
| LastName=user.LastName, | |||
| BirthDate=user.BirthDate, | |||
| GovernmentId=user.GovernmentId, | |||
| Address=user.Address, | |||
| ContactNumber=user.ContactNumber, | |||
| Role=user.Role, | |||
| HashedPassword=Hasher.get_password_hash(user.Password), | |||
| @@ -57,3 +58,20 @@ def get_car_driver(vehicle_id: int, db: Session): | |||
| def list_users(db: Session, role: str = "Any"): | |||
| users = db.query(User).filter((User.Role == role) | (role == "Any")).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: | |||
| return "userNotFound" | |||
| user.Email = user_data.Email | |||
| user.Name = user_data.Name | |||
| user.MiddleName = user_data.MiddleName | |||
| user.LastName = user_data.LastName | |||
| user.GovernmentId = user_data.GovernmentId | |||
| user.Address = user_data.Address | |||
| user.ContactNumber = user_data.ContactNumber | |||
| user.Role = user_data.Role | |||
| user.HashedPassword = Hasher.get_password_hash(user_data.Password) | |||
| db.commit() | |||
| db.refresh(user) | |||
| return user | |||
| @@ -10,8 +10,7 @@ def create_new_vehicle(vehicle: CreateVehicle, db: Session): | |||
| **vehicle.model_dump(), | |||
| Fuel=0, | |||
| Status="Inactive", | |||
| CurrentLocation=[], | |||
| MaintenanceNotes=[] | |||
| CurrentLocation=[] | |||
| ) | |||
| db.add(vehicle_object) | |||
| db.commit() | |||
| @@ -62,6 +61,8 @@ def list_vehicles(db: Session): | |||
| def get_vehicle_by_id(vehicle_id: int, db: Session): | |||
| vehicle = db.query(Vehicle).filter(Vehicle.Id == vehicle_id).first() | |||
| if not vehicle: | |||
| return None | |||
| driver = get_car_driver(vehicle.Id, db) | |||
| if driver: | |||
| vehicle.AssignedDriver = driver.Id | |||
| @@ -78,7 +79,8 @@ def replace_vehicle_data(id: int, vehicle: UpdateVehicle, db: Session): | |||
| vehicle_object.CurrentLocation = vehicle.CurrentLocation | |||
| vehicle_object.Fuel = vehicle.Fuel | |||
| vehicle_object.LicensePlate = vehicle.LicensePlate | |||
| vehicle_object.MaintenanceNotes = vehicle.MaintenanceNotes | |||
| vehicle_object.DriverHistory = vehicle.DriverHistory | |||
| vehicle_object.Type = vehicle.Type | |||
| vehicle_object.Mileage = vehicle.Mileage | |||
| vehicle_object.Model = vehicle.Model | |||
| vehicle_object.Status = vehicle.Status | |||
| @@ -27,8 +27,3 @@ app = startup() | |||
| @app.get("/") | |||
| def root(): | |||
| return {"message": "Hello World!"} | |||
| @app.get("/user") | |||
| def get_users(): | |||
| return {{"name": "almaz"}, {"name": "madi"}} | |||
| @@ -0,0 +1,10 @@ | |||
| from pydantic import BaseModel, Field | |||
| class CreateCarPart(BaseModel): | |||
| ParentId: int = Field(...) | |||
| Name: str = Field(...) | |||
| Number: str = Field(...) | |||
| Condition: str = Field(...) | |||
| ImageURL: str = Field(...) | |||
| Cost: int = Field(...) | |||
| @@ -0,0 +1,20 @@ | |||
| from pydantic import BaseModel, Field | |||
| class CreateTask(BaseModel): | |||
| DriverId: int = Field() | |||
| Description: str = Field(..., min_length=3, max_length=200) | |||
| StartLocation: tuple[str, str] = Field(...) | |||
| EndLocation: tuple[str, str] = Field(...) | |||
| class ShowTask(BaseModel): | |||
| Id: int | |||
| DriverId: int | |||
| Description: str | |||
| Status: str | |||
| StartLocation: tuple[str, str] | |||
| EndLocation: tuple[str, str] | |||
| class Config: | |||
| orm_mode = True | |||
| @@ -0,0 +1,7 @@ | |||
| from pydantic import BaseModel, Field | |||
| from datetime import datetime | |||
| class CreateMaintenanceJob(BaseModel): | |||
| Description: str = Field(...) | |||
| Date: datetime = Field(...) | |||
| @@ -1,6 +1,5 @@ | |||
| # Purpose: User schema for pydantic (validation, inside-api usage) | |||
| from datetime import datetime | |||
| from pydantic import BaseModel, EmailStr, Field | |||
| @@ -11,7 +10,8 @@ class UserCreate(BaseModel): | |||
| MiddleName: str = Field(None) | |||
| LastName: str = Field(..., min_length=3, max_length=50) | |||
| ContactNumber: str = Field(..., min_length=12, max_length=12) | |||
| BirthDate: datetime = Field(...) | |||
| GovernmentId: str = Field(...) | |||
| Address: str = Field(...) | |||
| Email: EmailStr = Field(...) | |||
| Role: str = Field(..., min_length=3, max_length=50) | |||
| @@ -22,7 +22,8 @@ class ShowUser(BaseModel): | |||
| MiddleName: str | None | |||
| LastName: str | |||
| ContactNumber: str | |||
| BirthDate: datetime | |||
| GovernmentId: str | |||
| Address: str | |||
| Email: EmailStr | |||
| Role: str | |||
| AssignedVehicle: int | None | |||
| @@ -30,3 +31,4 @@ class ShowUser(BaseModel): | |||
| class Config: | |||
| orm_mode = True | |||
| validate_assignment = True | |||
| @@ -8,6 +8,7 @@ class CreateVehicle(BaseModel): | |||
| LicensePlate: str | |||
| Mileage: int | |||
| Capacity: int | |||
| Type: str | |||
| class OutputVehicle(BaseModel): | |||
| @@ -18,23 +19,15 @@ class OutputVehicle(BaseModel): | |||
| Mileage: int | |||
| CurrentLocation: Optional[list[str]] = None | |||
| Fuel: Optional[int] = 0 | |||
| MaintenanceNotes: Optional[list[str]] = None | |||
| Type: str | |||
| DriverHistory: Optional[list[int]] = None | |||
| AssignedDriver: Optional[int] = None | |||
| Capacity: int | |||
| Status: str | |||
| class UpdateVehicle(BaseModel): | |||
| Model: str | |||
| Year: int | |||
| LicensePlate: str | |||
| Capacity: int | |||
| Mileage: int | |||
| Status: str | |||
| CurrentLocation: Optional[list[str]] = None | |||
| Fuel: Optional[int] = 0 | |||
| MaintenanceNotes: Optional[list[str]] = None | |||
| AssignedDriver: Optional[int] = None | |||
| class UpdateVehicle(OutputVehicle): | |||
| pass | |||
| class VehicleLocation(BaseModel): | |||