@@ -32,6 +32,10 @@ def getAuction( | |||
current_user: User = Depends(get_current_user), | |||
): | |||
auction = get_auction_by_id(id, db) | |||
if auction is None: | |||
raise HTTPException( | |||
status_code=404, detail="Auction with this ID does not exist" | |||
) | |||
return auction | |||
@@ -4,14 +4,21 @@ from db.session import get_db | |||
from db.repository.fuelingtask import ( | |||
create_fueling_task, | |||
delete_fueling_task, | |||
get_fueling_task_by_id | |||
get_fueling_task_by_id, | |||
get_all_fueling_tasks, | |||
) | |||
from schemas.fuelingtask import ( | |||
CreateFuelingTask, | |||
OutputFuelingTask, | |||
OutputFuelingTaskMin, | |||
OutputFuelingTaskList, | |||
) | |||
from schemas.fuelingtask import CreateFuelingTask, OutputFuelingTask | |||
from db.models.user import User | |||
from apis.v1.route_auth import get_current_user | |||
router = APIRouter() | |||
@router.post("/", response_model=OutputFuelingTask, status_code=status.HTTP_201_CREATED) | |||
def create_fuelingtask( | |||
fuelingtask: CreateFuelingTask = Depends(), | |||
@@ -28,19 +35,18 @@ def create_fuelingtask( | |||
) | |||
print("Created FuelTask") | |||
if fuelingtask_res == "nodriver": | |||
raise HTTPException( | |||
status_code=404, detail="Driver ID not found" | |||
) | |||
raise HTTPException(status_code=404, detail="Driver ID not found") | |||
if fuelingtask_res == "novehicle": | |||
raise HTTPException( | |||
status_code=404, detail="Vehicle ID not found" | |||
) | |||
raise HTTPException(status_code=404, detail="Vehicle ID not found") | |||
return fuelingtask_res | |||
@router.get("/{fueling_task_id}", response_model=OutputFuelingTask, status_code=status.HTTP_200_OK) | |||
@router.get( | |||
"/{fueling_task_id}", | |||
response_model=OutputFuelingTask, | |||
status_code=status.HTTP_200_OK, | |||
) | |||
def get_fuelingtask( | |||
fueling_task_id: int, | |||
db: Session = Depends(get_db), | |||
@@ -50,26 +56,35 @@ def get_fuelingtask( | |||
raise HTTPException( | |||
status_code=403, detail="You are not authorized to perform this action" | |||
) | |||
fuelingtask = get_fueling_task_by_id(fueling_task_id, db) | |||
if fuelingtask == "notfound": | |||
raise HTTPException( | |||
status_code=404, detail="fuck off" | |||
) | |||
raise HTTPException(status_code=404, detail="fuck off") | |||
return fuelingtask | |||
@router.delete("/{fueling_task_id}", status_code=status.HTTP_204_NO_CONTENT) | |||
def delete_fuelingtask( | |||
fueling_task_id: int, | |||
db: Session = Depends(get_db), | |||
current_user: User = Depends(get_current_user), | |||
): | |||
if current_user.Role != "FuelingPerson" and current_user.Role != "Admin": | |||
if current_user.Role != "Fueling" and current_user.Role != "Admin": | |||
raise HTTPException( | |||
status_code=403, detail="You are not authorized to perform this action" | |||
) | |||
if not delete_fueling_task(fueling_task_id, db): | |||
raise HTTPException(status_code=404, detail="Fueling task not found") | |||
@router.get("/", response_model=OutputFuelingTaskList, status_code=status.HTTP_200_OK) | |||
def get_all( | |||
db: Session = Depends(get_db), current_user: User = Depends(get_current_user) | |||
): | |||
if current_user.Role != "Admin" and current_user.Role != "Fueling": | |||
raise HTTPException( | |||
status_code=404, detail="Fueling task not found" | |||
) | |||
status_code=403, detail="You are not authorized to perform this action" | |||
) | |||
tasks = get_all_fueling_tasks(db) | |||
return tasks |
@@ -34,7 +34,9 @@ def create_maintenancejob( | |||
return maintenancejob_res | |||
@router.post("/carpart", response_model = ShowCarPart ,status_code=status.HTTP_201_CREATED) | |||
@router.post( | |||
"/carpart", response_model=ShowCarPart, status_code=status.HTTP_201_CREATED | |||
) | |||
def create_carpart( | |||
car_part: CreateCarPart = Depends(), | |||
db: Session = Depends(get_db), | |||
@@ -82,6 +84,10 @@ def get_maintenancejob( | |||
) | |||
maintenancejob = get_maintenance_job(maintenance_job_id, db) | |||
if maintenancejob is None: | |||
raise HTTPException( | |||
status_code=404, detail="Maintenance job with this id does not exist" | |||
) | |||
print(maintenancejob) | |||
return maintenancejob | |||
@@ -165,14 +165,16 @@ def getActiveRoute( | |||
return route | |||
@router.get("/myroutes", status_code=status.HTTP_200_OK) | |||
@router.get("/myroutes/", status_code=status.HTTP_200_OK) | |||
def getMyRoutes( | |||
db: Session = Depends(get_db), | |||
current_user: User = Depends(get_current_user), | |||
): | |||
print("here") | |||
if current_user.Role != "Driver": | |||
raise HTTPException( | |||
status_code=403, detail="You are not authorized to perform this action" | |||
status_code=403, | |||
detail="You are not a driver, you can't have routes, silly!", | |||
) | |||
routes = get_my_routes(current_user.Id, db) | |||
if routes == "notdriver": | |||
@@ -10,7 +10,7 @@ class FuelingTask(Base): | |||
CreatedById = Column(ForeignKey("user.Id"), nullable=False) | |||
Date = Column(DateTime, nullable=False) | |||
Description = Column(String, nullable=True) | |||
Cost = Column(Integer, nullable=False) | |||
Cost = Column(Float, nullable=False) | |||
FuelRefilled = Column(Float, nullable=False) | |||
GasStationName = Column(String, nullable=False) | |||
ImageBefore = Column(LargeBinary, nullable=False) | |||
@@ -14,6 +14,8 @@ def get_all_auctions(db: Session): | |||
def get_auction_by_id(id: int, db: Session): | |||
auction = db.query(Auction).filter(Auction.Id == id).first() | |||
if not auction: | |||
return None | |||
auction.car = auction.vehicle | |||
return auction | |||
@@ -63,10 +63,10 @@ def get_tasks_by_driver(driver_id: int, db: Session): | |||
def get_task_by_id(task_id: int, db: Session): | |||
task = db.query(DriveTask).filter(DriveTask.Id == task_id).first() | |||
task.Driver = task.CreatedBy.__dict__ | |||
task.Driver["AssignedVehicle"] = task.CreatedBy.vehicle | |||
if not task: | |||
return "notaskfound" | |||
task.Driver = task.CreatedBy.__dict__ | |||
task.Driver["AssignedVehicle"] = task.CreatedBy.vehicle | |||
return task | |||
@@ -7,16 +7,17 @@ from db.models.fuelingtask import FuelingTask | |||
from db.repository.user import get_car_driver, get_user_by_id | |||
from db.repository.vehicle import get_vehicle_by_id | |||
def create_fueling_task(fueling_task: CreateFuelingTask, current_user: int, db: Session): | |||
def create_fueling_task( | |||
fueling_task: CreateFuelingTask, current_user: int, db: Session | |||
): | |||
if not get_vehicle_by_id(fueling_task.VehicleId, db=db): | |||
return "novehicle" | |||
driver = get_car_driver(fueling_task.VehicleId, db=db) | |||
if not driver: | |||
return "nodriver" | |||
fueling_task_object = FuelingTask( | |||
DriverId=driver.Id, | |||
VehicleId=fueling_task.VehicleId, | |||
@@ -35,31 +36,59 @@ def create_fueling_task(fueling_task: CreateFuelingTask, current_user: int, db: | |||
print(driver.__dict__) | |||
driverobj = driver.__dict__ | |||
driverobj["AssignedVehicle"] = driver.vehicle.__dict__ | |||
fueling_task_object.Driver = driverobj | |||
return fueling_task_object | |||
resobj = fueling_task_object.__dict__ | |||
resobj["Driver"] = driverobj | |||
ia = fueling_task_object.ImageAfter | |||
ib = fueling_task_object.ImageBefore | |||
if ia is not None: | |||
ia = base64.b64encode(ia).decode("ascii") | |||
if ib is not None: | |||
ib = base64.b64encode(ib).decode("ascii") | |||
resobj["ImageBefore"] = ib | |||
resobj["ImageAfter"] = ia | |||
return resobj | |||
def delete_fueling_task(fueling_task_id: int, db: Session): | |||
fueling_task = db.query(FuelingTask).filter(FuelingTask.Id == fueling_task_id).first() | |||
fueling_task = ( | |||
db.query(FuelingTask).filter(FuelingTask.Id == fueling_task_id).first() | |||
) | |||
if fueling_task: | |||
db.delete(fueling_task) | |||
db.commit() | |||
return True | |||
return False | |||
def get_fueling_task_by_id(fuel_task_id: int, db: Session): | |||
fuel_task = db.query(FuelingTask).filter(FuelingTask.Id == fuel_task_id).first() | |||
if not fuel_task: | |||
return "notfound" | |||
res = fuel_task.__dict__ | |||
res = fuel_task.__dict__ | |||
driver = get_user_by_id(fuel_task.DriverId, role="Driver", db=db) | |||
driver_obj = driver.__dict__ | |||
imagebefore = fuel_task.ImageBefore | |||
imageafter = fuel_task.ImageAfter | |||
imagebeforeBase64 = base64.b64encode(imagebefore).decode('ascii') | |||
imageafterBase64 = base64.b64encode(imageafter).decode('ascii') | |||
imagebeforeBase64 = base64.b64encode(imagebefore).decode("ascii") | |||
imageafterBase64 = base64.b64encode(imageafter).decode("ascii") | |||
res["ImageBefore"] = imagebeforeBase64 | |||
res["ImageAfter"] = imageafterBase64 | |||
res["Driver"] = driver_obj | |||
res["Driver"]["AssignedVehicle"] = driver.vehicle | |||
return res | |||
return res | |||
def get_all_fueling_tasks(db: Session): | |||
fuel_task = db.query(FuelingTask).all() | |||
result = [] | |||
for task in fuel_task: | |||
res = task.__dict__ | |||
res["ImageBefore"] = "" | |||
res["ImageAfter"] = "" | |||
driver = get_user_by_id(task.DriverId, role="Driver", db=db) | |||
driver_obj = driver.__dict__ | |||
res["Driver"] = driver_obj | |||
res["Driver"]["AssignedVehicle"] = driver.vehicle | |||
result.append(res) | |||
x = {"FuelingTasks": result} | |||
return x |
@@ -53,16 +53,23 @@ def calculate_total_cost(car_parts: CarPart): | |||
def get_all_maintenance_jobs(db: Session): | |||
maintenancejobs = db.query(MaintenanceJob).all() | |||
print("DB Access complete") | |||
result = [] | |||
for job in maintenancejobs: | |||
job_dict = job.__dict__ | |||
print(job_dict) | |||
job_dict["CarPartsList"] = [part.__dict__ for part in job.CarParts] | |||
print(len(job_dict["CarPartsList"])) | |||
for part in job_dict["CarPartsList"]: | |||
part["image"] = base64.b64encode(part["ImageURL"]).decode("ascii") | |||
if (part["ImageURL"] is None) or (part["ImageURL"] == ""): | |||
part["image"] = "" | |||
else: | |||
part["image"] = base64.b64encode(part["ImageURL"]).decode("ascii") | |||
job_dict["TotalCost"] = calculate_total_cost(job.CarParts) | |||
job_dict["AssignedTo"] = job.CreatedBy.__dict__ | |||
job_dict["Vehicle"] = job.Vehicle.__dict__ | |||
result.append(job_dict) | |||
print("Returning...") | |||
return maintenancejobs | |||
@@ -70,11 +77,13 @@ def get_maintenance_job(maintenancejob_id: int, db: Session): | |||
maintenancejob = ( | |||
db.query(MaintenanceJob).filter(MaintenanceJob.Id == maintenancejob_id).first() | |||
) | |||
if maintenancejob is None: | |||
return None | |||
res = maintenancejob.__dict__ | |||
res["CarPartsList"] = [part.__dict__ for part in maintenancejob.CarParts] | |||
for part in maintenancejob.CarPartsList: | |||
part["image"] = base64.b64encode(part["ImageURL"]).decode("ascii") | |||
# print(type(result.CarPartsList)) | |||
res["TotalCost"] = calculate_total_cost(maintenancejob.CarParts) | |||
# print(result.TotalCost) | |||
@@ -1,4 +1,3 @@ | |||
from fastapi import Form, UploadFile | |||
from pydantic import BaseModel, Field | |||
from dataclasses import dataclass | |||
@@ -9,10 +8,8 @@ class CreateCarPart(BaseModel): | |||
Name: str = Form(...) | |||
Number: str = Form(...) | |||
Condition: str = Form(...) | |||
Cost: int = Form(...) | |||
Cost: float = Form(...) | |||
image: UploadFile = Form(...) | |||
class ShowCarPart(BaseModel): | |||
@@ -20,5 +17,5 @@ class ShowCarPart(BaseModel): | |||
Name: str = Field(...) | |||
Number: str = Field(...) | |||
Condition: str = Field(...) | |||
Cost: int = Field(...) | |||
image: str = Field(...) | |||
Cost: float = Field(...) | |||
image: str = Field(...) |
@@ -5,7 +5,7 @@ from typing import Optional | |||
class CreateTask(BaseModel): | |||
DriverId: int = Field() | |||
DriverId: int = Field(...) | |||
Description: str = Field(..., min_length=3, max_length=200) | |||
StartLocation: tuple[str, str] = Field(...) | |||
EndLocation: tuple[str, str] = Field(...) | |||
@@ -3,29 +3,43 @@ from fastapi import Form, UploadFile | |||
from datetime import datetime | |||
from schemas.user import ShowDriver | |||
class CreateFuelingTask(BaseModel): | |||
VehicleId: int = Form(...) | |||
Description: str = Form(...) | |||
Date: datetime = Form(...) | |||
Cost: int = Form(...) | |||
FuelRefilled: int = Form(...) | |||
Cost: float = Form(...) | |||
FuelRefilled: float = Form(...) | |||
GasStationName: str = Form(...) | |||
ImageBefore: UploadFile = Form(...) | |||
ImageAfter: UploadFile = Form(...) | |||
model_config={ | |||
"arbitrary_types_allowed": True | |||
} | |||
model_config = {"arbitrary_types_allowed": True} | |||
class OutputFuelingTask(BaseModel): | |||
Id: int = Field(...) | |||
VehicleId: int = Field(...) | |||
Description: str = Field(...) | |||
Date: datetime = Field(...) | |||
Cost: int = Field(...) | |||
FuelRefilled: int = Field(...) | |||
Cost: float = Field(...) | |||
FuelRefilled: float = Field(...) | |||
GasStationName: str = Field(...) | |||
Driver: ShowDriver | None | |||
ImageBefore: str = Field(...) | |||
ImageAfter: str = Field(...) | |||
model_config={ | |||
"arbitrary_types_allowed": True | |||
} | |||
model_config = {"arbitrary_types_allowed": True} | |||
class OutputFuelingTaskMin(BaseModel): | |||
Id: int = Field(...) | |||
VehicleId: int = Field(...) | |||
Description: str = Field(...) | |||
Date: datetime = Field(...) | |||
Cost: float = Field(...) | |||
FuelRefilled: float = Field(...) | |||
GasStationName: str = Field(...) | |||
Driver: ShowDriver | None | |||
class OutputFuelingTaskList(BaseModel): | |||
FuelingTasks: list[OutputFuelingTaskMin] |
@@ -18,6 +18,6 @@ class OutputMaintenanceJob(BaseModel): | |||
Description: str | |||
Date: datetime | |||
CarPartsList: Optional[List[ShowCarPart]] | |||
TotalCost: int | |||
TotalCost: float | |||
Vehicle: OutputVehicle | |||
AssignedTo: ShowUser |