Add endpoint for creating inventory dewar

Author: gfrn

src/scaup/crud/internal.py

@@ -1,11 +1,12 @@
-from sqlalchemy import select
+from sqlalchemy import insert, select
 from sqlalchemy.orm import joinedload
 
 from ..models.inner_db.tables import Container, TopLevelContainer
 from ..models.shipments import ShipmentChildren
 from ..models.top_level_containers import TopLevelContainerOut
 from ..utils.database import inner_db
 from ..utils.query import query_result_to_object
+from ..utils.session import retry_if_exists
 
 
 def get_unassigned(limit: int, page: int):
@@ -43,3 +44,39 @@ def get_internal_containers(limit: int, page: int):
     query = select(TopLevelContainer).filter(TopLevelContainer.isInternal.is_(True))
 
     return inner_db.paginate(query, limit, page, slow_count=False, scalar=False)
+
+
+@retry_if_exists
+def create_preloaded_inventory_dewar(name: str):
+    tlc = inner_db.session.scalar(
+        insert(TopLevelContainer).returning(TopLevelContainer),
+        {"name": name, "isInternal": True, "code": name},
+    )
+
+    containers = inner_db.session.scalars(
+        insert(Container).returning(Container),
+        [
+            {"name": i, "isInternal": True, "topLevelContainerId": tlc.id, "type": "puck", "subType": "2"}
+            for i in range(1, 6)
+        ],
+    ).all()
+
+    inner_db.session.execute(
+        insert(Container),
+        [
+            {
+                "name": f"Gridbox_{pos + 1}_Puck_{container.name}",
+                "isInternal": True,
+                "parentId": container.id,
+                "type": "gridBox",
+                "subType": "auto",
+                "location": pos + 1,
+            }
+            for pos in range(12)
+            for container in containers
+        ],
+    )
+
+    inner_db.session.commit()
+
+    return tlc

src/scaup/models/containers.py

@@ -68,7 +68,7 @@ class OptionalContainer(BaseContainer):
 
 
 class ContainerOut(BaseContainer):
-    shipmentId: int
+    shipmentId: int | None = None
     id: int = Field(validation_alias=AliasChoices("containerId", "id"))
     model_config = ConfigDict(from_attributes=True, arbitrary_types_allowed=True)
     type: str

src/scaup/models/samples.py

@@ -40,7 +40,7 @@ class OptionalSample(BaseSample):
 
 class SampleOut(BaseSample):
     id: int = Field(validation_alias=AliasChoices("sampleId", "id", AliasPath("Sample", "id")))
-    shipmentId: int = Field(validation_alias=InnerAlias("shipmentId"))
+    shipmentId: int | None = Field(default=None, validation_alias=InnerAlias("shipmentId"))
     proteinId: int = Field(validation_alias=InnerAlias("proteinId"))
     containerName: Optional[str] = None
     type: str = Field(validation_alias=InnerAlias("type"))

src/scaup/models/top_level_containers.py

@@ -14,6 +14,10 @@ class TopLevelContainerHistory(BaseModel):
     arrivalDate: datetime | None = None
 
 
+class PreloadedInventoryDewar(BaseModel):
+    name: str
+
+
 class BaseTopLevelContainer(BaseModel):
     details: Optional[dict[str, Any]] = None
     comments: Optional[str] = None

src/scaup/routes/internal.py

@@ -8,7 +8,7 @@
 from ..crud.top_level_containers import create_top_level_container
 from ..models.containers import ContainerIn, ContainerOut
 from ..models.shipments import GenericItem, ShipmentChildren
-from ..models.top_level_containers import TopLevelContainerIn, TopLevelContainerOut
+from ..models.top_level_containers import PreloadedInventoryDewar, TopLevelContainerIn, TopLevelContainerOut
 from ..utils.auth import check_em_staff
 
 
@@ -77,3 +77,13 @@ def create_orphan_top_level_container(
 def get_internal_container(topLevelContainerId: int):
     """Get internal top level container and its children"""
     return crud.get_internal_container_tree(top_level_container_id=topLevelContainerId)
+
+
+@router.post(
+    "/preloaded-dewars",
+    response_model=TopLevelContainerOut,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_preloaded_inventory_dewar(parameters: PreloadedInventoryDewar = Body()):
+    """Create preloaded inventory dewar with pucks and grid boxes"""
+    return crud.create_preloaded_inventory_dewar(name=parameters.name)

tests/internal/test_create_preloaded.py

@@ -0,0 +1,25 @@
+import pytest
+from sqlalchemy import func, select
+
+from scaup.models.inner_db.tables import Container, TopLevelContainer
+from scaup.utils.database import inner_db
+
+from ..test_utils.users import admin
+
+
+@pytest.mark.parametrize("mock_user", [admin], indirect=True)
+def test_create(mock_user, client):
+    """Should create internal preloaded dewar"""
+    resp = client.post(
+        "/internal-containers/preloaded-dewars",
+        json={"name": "foo"},
+    )
+
+    assert resp.status_code == 201
+
+    item_id = resp.json()["id"]
+
+    assert inner_db.session.scalar(select(TopLevelContainer).filter(TopLevelContainer.id == item_id)) is not None
+    assert (
+        inner_db.session.scalar(select(func.count(Container.id)).filter(Container.topLevelContainerId == item_id)) == 5
+    )