Add tags to ingest process

Author: CannonLock

api/models/ingest.py

@@ -2,9 +2,9 @@
 from typing import Optional
 from enum import Enum
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, field_validator
 
-from api.schemas import IngestState
+from api.schemas import IngestState, IngestProcessTag
 from .source import Sources
 
 
@@ -15,6 +15,7 @@ class Post(BaseModel):
     source_id: Optional[int] = None
     access_group_id: Optional[int] = None
     map_id: Optional[str] = None
+    tags: Optional[list[str]] = None
 
     class Config:
         orm_mode = True
@@ -29,7 +30,22 @@ class Get(Post):
     completed_on: Optional[datetime.datetime] = None
     source: Optional[Sources] = None
 
+    @field_validator("tags", mode="before")
+    @classmethod
+    def transform_tags(cls, v):
+        if len(v) == 0:
+            return []
+
+        if isinstance(v[0], IngestProcessTag):
+            return [tag.tag for tag in v]
+
+        return v
+
 
 class Patch(Post):
     pass
 
+
+class Tag(BaseModel):
+    tag: str
+

api/routes/ingest.py

@@ -1,7 +1,7 @@
 import os
 
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy import insert, select, update, and_
+from sqlalchemy import insert, select, update, and_, delete
 from sqlalchemy.orm import selectinload, joinedload, defer
 import minio
 
@@ -13,7 +13,7 @@
 from api.routes.security import has_access
 import api.models.ingest as IngestProcessModel
 import api.models.object as Object
-from api.schemas import IngestProcess as IngestProcessSchema, ObjectGroup, Sources
+from api.schemas import IngestProcess as IngestProcessSchema, ObjectGroup, Sources, IngestProcessTag
 from api.query_parser import get_filter_query_params, QueryParser
 
 router = APIRouter(
@@ -39,7 +39,8 @@ async def get_multiple_ingest_process(page: int = 0, page_size: int = 50, filter
             .limit(page_size)\
             .offset(page_size * page)\
             .where(and_(query_parser.where_expressions()))\
-            .options(joinedload(IngestProcessSchema.source).defer(Sources.rgeom).defer(Sources.web_geom))
+            .options(joinedload(IngestProcessSchema.source).defer(Sources.rgeom).defer(Sources.web_geom))\
+            .options(selectinload(IngestProcessSchema.tags))
 
         results = await session.execute(select_stmt)
 
@@ -56,7 +57,8 @@ async def get_ingest_process(id: int):
     async with async_session() as session:
 
         select_stmt = select(IngestProcessSchema).where(and_(IngestProcessSchema.id == id))\
-            .options(joinedload(IngestProcessSchema.source).defer(Sources.rgeom).defer(Sources.web_geom))
+            .options(joinedload(IngestProcessSchema.source).defer(Sources.rgeom).defer(Sources.web_geom))\
+            .options(selectinload(IngestProcessSchema.tags))
 
         result = await session.scalar(select_stmt)
 
@@ -78,18 +80,25 @@ async def create_ingest_process(object: IngestProcessModel.Post, user_has_access
 
     async with async_session() as session:
 
-        object_group_stmt = insert(ObjectGroup).values().returning(ObjectGroup)
-        object_group = await session.scalar(object_group_stmt)
-
-        stmt = insert(IngestProcessSchema).values(**object.model_dump(), object_group_id=object_group.id).returning(IngestProcessSchema)
+        object_group = ObjectGroup()
+        session.add(object_group)
+        await session.commit()
 
-        server_object = await session.scalar(stmt)
+        tags = [IngestProcessTag(tag=tag) for tag in object.tags]
+        del object.tags
 
-        server_object.source = await session.scalar(select(Sources).where(Sources.source_id == server_object.source_id))
+        ingest_process = IngestProcessSchema(
+            **object.model_dump(),
+            object_group_id=object_group.id,
+            tags=tags
+        )
 
+        session.add(ingest_process)
         await session.commit()
 
-    return server_object
+        ingest_process.source = await session.get(Sources, object.source_id)
+
+    return ingest_process
 
 
 @router.patch("/{id}", response_model=IngestProcessModel.Get)
@@ -119,6 +128,62 @@ async def patch_ingest_process(
         await session.commit()
         return response
 
+@router.post("/{id}/tags", response_model=list[str])
+async def add_ingest_process_tag(
+    id: int,
+    tag: IngestProcessModel.Tag,
+    user_has_access: bool = Depends(has_access)
+):
+    """Add a tag to an ingest process"""
+
+    if not user_has_access:
+        raise HTTPException(status_code=403, detail="User does not have access to create an object")
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        ingest_process = await session.get(IngestProcessSchema, id)
+
+        if ingest_process is None:
+            raise HTTPException(status_code=404, detail=f"IngestProcess with id ({id}) not found")
+
+        ingest_process.tags.append(IngestProcessTag(tag=tag.tag))
+        await session.commit()
+
+        ingest_process = await session.get(IngestProcessSchema, id)
+        return [tag.tag for tag in ingest_process.tags]
+
+    return None
+
+@router.delete("/{id}/tags/{tag}", response_model=list[str])
+async def delete_ingest_process_tag(id: int, tag: str, user_has_access: bool = Depends(has_access)):
+    """Delete a tag from an ingest process"""
+
+    if not user_has_access:
+        raise HTTPException(status_code=403, detail="User does not have access to create an object")
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        ingest_process = await session.get(IngestProcessSchema, id)
+
+        if ingest_process is None:
+            raise HTTPException(status_code=404, detail=f"IngestProcess with id ({id}) not found")
+
+        tag_stmt = delete(IngestProcessTag).where(and_(IngestProcessTag.ingest_process_id == id, IngestProcessTag.tag == tag))
+        await session.execute(tag_stmt)
+        await session.commit()
+
+        ingest_process = await session.get(IngestProcessSchema, id)
+
+        return [tag.tag for tag in ingest_process.tags]
+
+    return ingest_process
+
 
 @router.get("/{id}/objects", response_model=list[Object.GetSecureURL])
 async def get_ingest_process_objects(id: int):

api/schemas.py

@@ -1,7 +1,7 @@
 import enum
 from typing import List
 import datetime
-from sqlalchemy import ForeignKey, func, DateTime, Enum, UniqueConstraint
+from sqlalchemy import ForeignKey, func, DateTime, Enum, PrimaryKeyConstraint, UniqueConstraint
 from sqlalchemy.dialects.postgresql import VARCHAR, TEXT, INTEGER, ARRAY, BOOLEAN, JSON, JSONB
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from geoalchemy2 import Geometry
@@ -171,3 +171,18 @@ class IngestProcess(Base):
     # Relationships
     object_group: Mapped[ObjectGroup] = relationship(back_populates="ingest_process", lazy="joined")
     source: Mapped[Sources] = relationship(back_populates="ingest_process")
+    tags: Mapped[List["IngestProcessTag"]] = relationship(back_populates="ingest_process", lazy="joined")
+
+
+class IngestProcessTag(Base):
+    __tablename__ = "ingest_process_tag"
+    __table_args__ = (
+        PrimaryKeyConstraint('ingest_process_id', 'tag', name='pk_tag'),
+        {'schema': 'maps_metadata'}
+    )
+
+    ingest_process_id: Mapped[int] = mapped_column(ForeignKey("maps_metadata.ingest_process.id"))
+    tag: Mapped[str] = mapped_column(VARCHAR(255))
+
+    # Relationships
+    ingest_process: Mapped[IngestProcess] = relationship(back_populates="tags")

api/tests/main.py

@@ -433,7 +433,8 @@ def test_add_ingest_process(self, api_client):
 
     ingest_process_data = {
       "comments": "This is a test comment",
-      "state": "pending"
+      "state": "pending",
+      "tags" : ["test_tag", "cool_tag"]
     }
 
     response = api_client.post(
@@ -443,6 +444,12 @@ def test_add_ingest_process(self, api_client):
 
     assert response.status_code == 200
 
+    data = response.json()
+
+    assert data['comments'] == "This is a test comment"
+    assert data['state'] == "pending"
+    assert data['tags'] == ["test_tag", "cool_tag"]
+
   def test_get_ingest_processes(self, api_client):
     response = api_client.get("/ingest-process")
     assert response.status_code == 200
@@ -483,6 +490,52 @@ def test_patch_ingest_process(self, api_client):
 
     assert single_data['comments'] == "test"
 
+  def test_add_tag_to_ingest_process(self, api_client):
+    """Test adding a tag to an ingest process"""
+
+    test_tag = f"new_tag-{random.randint(0,10000000)}"
+
+    response = api_client.get("/ingest-process")
+    assert response.status_code == 200
+
+    data = response.json()
+
+    assert len(data) > 0
+
+    response = api_client.post(f"/ingest-process/{data[0]['id']}/tags", json={"tag": test_tag})
+
+    assert response.status_code == 200
+
+    single_data = response.json()
+
+    assert test_tag in single_data
+
+  def test_delete_tag_from_ingest_process(self, api_client):
+    """Test deleting a tag from an ingest process"""
+
+    test_tag = f"new_tag-{random.randint(0,10000000)}"
+
+    response = api_client.get("/ingest-process")
+    assert response.status_code == 200
+
+    data = response.json()
+
+    assert len(data) > 0
+
+    response = api_client.post(f"/ingest-process/{data[0]['id']}/tags", json={"tag": test_tag})
+    post_data = response.json()
+
+    assert response.status_code == 200
+    assert test_tag in post_data
+
+    response = api_client.delete(f"/ingest-process/{data[0]['id']}/tags/{test_tag}")
+
+    assert response.status_code == 200
+
+    single_data = response.json()
+
+    assert test_tag not in single_data
+
   def test_pair_object_to_ingest(self, api_client):
     response = api_client.get("/ingest-process")
     assert response.status_code == 200