updated production-run

Author: nameless-coder

src/database/connection.py

@@ -25,8 +25,10 @@ def __init__(self, model):
         self.model = model
 
     async def save(self, document):
-        await document.create()
-        return
+        doc = await document.create()
+        if doc:
+            return doc
+        return False
 
     async def get(self, id: PydanticObjectId) -> bool:
         doc = await self.model.get(id)

src/main.py

@@ -9,7 +9,7 @@
 
 settings = Settings()
 
-api.include_router(production_run_router)
+api.include_router(production_run_router, prefix="/productionrun")
 
 
 @api.on_event("startup")

src/models/production_run.py

@@ -1,29 +1,38 @@
+from typing import Optional
 from datetime import datetime
+
 from beanie import Document
 from pydantic import BaseModel
 
 
 class ProductionRun(Document):
-    id: int
+    pr_id: str
     name: str
-    dt: datetime = None
+    description: Optional[str]
+    started: datetime = None
 
     class Config:
         schema_extra = {
             "example": {
-                "id": 1,
-                "name": "Example Schema!",
-                "dt": '2022-12-23T10:20:30.400+02:00'
+                "pr_id": "FA-01",
+                "name": "1st production run",
+                "description": "Testing the equipment!",
+                "started": "2022-12-23T10:20:30.400+02:00"
             }
         }
 
+    class Settings:
+        name = "production_runs"
 
-class ProductionRunItem(BaseModel):
-    name: str
+
+class ProductionRunUpdate(BaseModel):
+    name: Optional[str]
+    description: Optional[str]
 
     class Config:
         schema_extra = {
             "example": {
-                "name": "Framing material"
+                "name": "Framing material",
+                "description": "Changed gears"
             }
         }

src/requirements.txt

@@ -12,6 +12,7 @@ passlib==1.7.4
 pydantic==1.10.2
 pymongo==4.3.3
 python-dotenv==0.21.0
+pytz==2022.7
 sniffio==1.3.0
 starlette==0.22.0
 typing_extensions==4.4.0

src/routes/production_run.py

@@ -1,69 +1,68 @@
-from fastapi import APIRouter, Path
+from typing import List
+from datetime import datetime
+import pytz
 
-from models.production_run import ProductionRun, ProductionRunItem
+from beanie import PydanticObjectId
+from database.connection import Database
+from fastapi import APIRouter, Path, HTTPException, status
+from models.production_run import ProductionRun, ProductionRunUpdate
 
-production_run_router = APIRouter()
+production_run_router = APIRouter(
+    tags=["ProductionRun"]
+)
 
 production_run_list = []
+production_run_database = Database(ProductionRun)
 
 
-@production_run_router.post("/productionrun")
-async def add_item(item: ProductionRun) -> dict:
-    production_run_list.append(item)
-    return {
-        "message": "Production run added successfully."
-    }
-
-
-@production_run_router.get("/productionrun")
-async def retrieve_item() -> dict:
-    return {
-        "productions_runs": production_run_list
-    }
+@production_run_router.get("/", response_model=List[ProductionRun])
+async def retrieve_all_production_runs() -> List[ProductionRun]:
+    production_runs = await production_run_database.get_all()
+    return production_runs
 
 
-@production_run_router.get("/productionrun/{item_id}")
-async def get_single_item(item_id: int = Path(..., title="The ID of the item to retrieve.")) -> dict:
-    for item in production_run_list:
-        if item.id == item_id:
-            return {
-                "item": item
-            }
-    return {
-        "message": "Item with supplied ID doesn't exist."
-    }
+@production_run_router.get("/{item_id}", response_model=ProductionRun)
+async def retrieve_single_production_run(item_id: PydanticObjectId = Path(..., title="The ID of the item to retrieve.")) -> ProductionRun:
+    production_run = await production_run_database.get(item_id)
+    if not production_run:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Production run with supplied ID does not exist"
+        )
+    return production_run
 
 
-@production_run_router.put("/productionrun/{item_id}")
-async def update_item(item_data: ProductionRunItem, item_id: int = Path(..., title="The ID of the item to be updated.")) -> dict:
-    for item in production_run_list:
-        if item.id == item_id:
-            item.name = item_data.name
-            return {
-                "message": "Item updated successfully."
-            }
+@production_run_router.post("/new")
+async def create_production_run(body: ProductionRun) -> dict:
+    body.started = datetime.now(pytz.timezone('Europe/Riga'))
+    new_production_run = await production_run_database.save(body)
     return {
-        "message": "Item with supplied ID doesn't exist."
+        "message": "Production run created successfully.",
+        "pr": new_production_run
     }
 
 
-@production_run_router.delete("/productionrun/{item_id}")
-async def delete_single_item(item_id: int) -> dict:
-    for index in range(len(production_run_list)):
-        item = production_run_list[index]
-        if item.id == item_id:
-            production_run_list.pop(index)
-            return {
-                "message": "Item deleted successfully."
-            }
-    return {
-        "message": "Item with supplied ID doesn't exist."
-    }
+@production_run_router.put("/{item_id}", response_model=ProductionRun)
+async def update_production_run(item_data: ProductionRunUpdate,
+                                item_id: PydanticObjectId = Path(..., title="The ID of the item to be updated.")) -> ProductionRun:
+    updated_production_run = await production_run_database.update(item_id, item_data)
+    if not updated_production_run:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Production run with supplied ID does not exist"
+        )
+    return updated_production_run
 
 
-@production_run_router.delete("/productionrun")
-async def delete_all_items() -> dict:
-    production_run_list.clear()
+@production_run_router.delete("/{item_id}")
+async def delete_single_production_run(item_id: PydanticObjectId = Path(..., title="The ID of the item to be deleted.")) -> dict:
+    production_run = await production_run_database.get(item_id)
+    if not production_run:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Production run with supplied ID does not exist"
+        )
+    await production_run_database.delete(item_id)
     return {
-        "message": "Items deleted successfully."
+        "message": "Production run deleted successfully."
     }

src/test_main.http

@@ -4,3 +4,25 @@ GET http://127.0.0.1:8000/
 Accept: application/json
 
 ###
+
+GET http://127.0.0.1:8000/productionrun
+Accept: application/json
+
+###
+
+POST http://127.0.0.1:8000/productionrun/new
+Content-Type: application/json
+
+{
+  "pr_id": "FA-01",
+  "name": "1st production run",
+  "description": "Testing the equipment!"
+}
+
+###
+
+GET http://127.0.0.1:8000/productionrun
+Accept: application/json
+
+###
+