Add Source Form

api/models/ingest.py

@@ -5,7 +5,7 @@
 from pydantic import BaseModel, ConfigDict, field_validator
 
 from api.schemas import IngestState, IngestProcessTag
-from .source import Sources
+import api.models.source as Sources
 
 
 class Post(BaseModel):
@@ -26,7 +26,7 @@ class Get(Post):
     object_group_id: int
     created_on: datetime.datetime
     completed_on: Optional[datetime.datetime] = None
-    source: Optional[Sources] = None
+    source: Optional[Sources.Get] = None
 
     @field_validator("tags", mode="before")
     @classmethod

api/models/source.py

@@ -7,10 +7,9 @@
 
 
 # Database Models
-class Sources(BaseModel):
-    source_id: int
+class Post(BaseModel):
     name: Optional[str] = None
-    primary_table: str
+    primary_table: Optional[str] = None
     url: Optional[str] = None
     raster_url: Optional[str] = None
     ref_title: Optional[str] = None
@@ -23,7 +22,15 @@ class Sources(BaseModel):
     licence: Optional[str] = None
     features: Optional[int] = None
     area: Optional[int] = None
-    priority: bool
+    priority: Optional[bool] = None
     display_scales: Optional[list[str]] = None
-    new_priority: int
-    status_code: str
+    new_priority: Optional[int] = None
+    status_code: Optional[str] = None
+
+
+class Get(Post):
+    source_id: int
+
+
+class Patch(Post):
+    pass

api/routes/sources.py

@@ -23,7 +23,7 @@
 from api.models.geometries import (
     PolygonModel, PolygonRequestModel, PolygonResponseModel, CopyColumnRequest, LineStringModel, PointModel
 )
-from api.models.source import Sources
+import api.models.source as Sources
 from api.query_parser import ParserException
 from api.routes.security import has_access
 
@@ -42,7 +42,7 @@
 
 @router.get("")
 async def get_sources(response: Response, page: int = 0, page_size: int = 100, include_geom: bool = False) -> List[
-    Sources]:
+    Sources.Get]:
     async_session = get_async_session(get_engine())
     sources = await db.get_sources(async_session, page, page_size)
 
@@ -62,7 +62,7 @@ async def get_sources(response: Response, page: int = 0, page_size: int = 100, i
 
 
 @router.get("/{source_id}")
-async def get_source(source_id: int) -> Sources:
+async def get_source(source_id: int) -> Sources.Get:
     """Get a single object"""
 
     engine = get_engine()
@@ -78,20 +78,24 @@ async def get_source(source_id: int) -> Sources:
         if results is None:
             raise HTTPException(status_code=404, detail=f"Object with id ({id}) not found")
 
-        return db.results_to_model(results, Sources)[0]
+        return db.results_to_model(results, Sources.Get)[0]
 
 
 @router.patch("/{source_id}")
-async def patch_source(source_id: int, source: Sources, user_has_access: bool = Depends(has_access)) -> Sources:
+async def patch_source(source_id: int, source: Sources.Patch, user_has_access: bool = Depends(has_access)) -> Sources.Get:
     """Patch a source"""
 
+    if not user_has_access:
+        raise HTTPException(status_code=401, detail="User does not have access to patch object")
+
     engine = get_engine()
     async_session = get_async_session(engine)
 
     async with async_session() as session:
-        update_stmt = update(schemas.Sources) \
-            .where(schemas.Sources.source_id == source_id) \
-            .values(**source.model_dump(exclude_none=True))
+        update_stmt = update(schemas.Sources)\
+            .where(schemas.Sources.source_id == source_id)\
+            .values(**source.model_dump(exclude_none=True))\
+            .returning(schemas.Sources)
 
         server_object = await session.scalar(update_stmt)
 

api/tests/sources.py

@@ -20,6 +20,22 @@ def test_get_source(self, api_client: TestClient):
         assert response_json["source_id"] == TEST_SOURCE_TABLE.source_id
 
     def test_patch_source(self, api_client: TestClient):
+        response = api_client.patch(
+            f"/sources/{TEST_SOURCE_TABLE.source_id}",
+            json={
+                "url": "test"
+            }
+        )
+
+        assert response.status_code == 200
+
+        response = api_client.get(f"/sources/{TEST_SOURCE_TABLE.source_id}")
+
+        assert response.status_code == 200
+
+        response_json = response.json()
+
+        assert response_json["url"] == "test"
 
     def test_get_sub_source_geometries(self, api_client: TestClient):
         response = api_client.get(f"/sources/{1}/geometries")