assignments added (WIP)

app/apis/base.py

@@ -1,9 +1,10 @@
 # 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
 
 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"])

app/apis/v1/route_task.py

@@ -0,0 +1,32 @@
+from fastapi import Depends, APIRouter
+from sqlalchemy.orm import Session
+from fastapi import status, HTTPException
+from typing import Annotated
+from db.session import get_db
+from core.config import settings
+from db.repository.drivetask import (
+    create_new_task,
+)
+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

app/db/models/drivetask.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, Integer, String, ForeignKey, ARRAY
+from db.base import Base
+
+
+class DriveTask(Base):
+    Id = Column(Integer, primary_key=True, index=True)
+    DriverId = Column(ForeignKey("user.Id"), nullable=False)
+    Description = Column(String, nullable=True)
+    Status = Column(String, nullable=False)
+    StartLocation = Column(ARRAY(String), nullable=False)
+    EndLocation = Column(ARRAY(String), nullable=False)

app/db/repository/drivetask.py

@@ -0,0 +1,24 @@
+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

app/schemas/drivetask.py

@@ -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