Add Object Endpoints

- Add the object endpoints
- Refactor some existing code to account for larger code base

Author: CannonLock

.github/workflows/build-dev.yaml

@@ -2,6 +2,7 @@ name: Build Development
 
 on:
   push:
+    branches: [ '**' ]
     tags:
       - v[0-9]+.[0-9]+.[0-9]+-**  # Semver Pre-Release
   pull_request:

.github/workflows/build-prod.yaml

@@ -23,6 +23,7 @@ jobs:
         with:
           images: hub.opensciencegrid.org/macrostrat/api-v3
           tags: |
+            type=ref,event=pr,suffix=-{{date 'YYYYMMDDHHmmss'}}
             type=ref,event=branch,suffix=-{{date 'YYYYMMDDHHmmss'}}
             type=semver,pattern={{version}}
             type=raw,value=latest,enable={{is_default_branch}}

README.md

@@ -0,0 +1,6 @@
+# Macrostrat API V3
+
+## Overview
+
+This is a Fastapi application interfacing with a postgres database. It is designed to be deployed behind 
+Nginx on a kubernetes cluster.

api/app.py

@@ -24,9 +24,10 @@
     patch_sources_sub_table,
     select_sources_sub_table,
 )
-from api.models import PolygonModel, Sources, CopyColumnRequest
+from api.models.source import PolygonModel, Sources, CopyColumnRequest
 from api.query_parser import ParserException
 from api.routes.security import TokenData, get_groups
+from api.routes.object import router as object_router
 
 @asynccontextmanager
 async def setup_engine(a: FastAPI):
@@ -56,6 +57,7 @@ async def setup_engine(a: FastAPI):
 )
 
 app.include_router(api.routes.security.router)
+app.include_router(object_router)
 
 
 @app.get("/sources")
@@ -172,5 +174,8 @@ async def patch_sub_sources(
 
 
 
+
+
+
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8000, headers=[("Access-Control-Expose-Headers", "X-Total-Count")])

api/database.py

@@ -7,8 +7,9 @@
 #
 import datetime
 from os import environ
-from typing import Type
+from typing import Type, List
 
+from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
 from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
 from sqlalchemy import text, select, update, Table, MetaData, CursorResult, func, insert
@@ -134,6 +135,15 @@ async def get_access_token(async_session: async_sessionmaker[AsyncSession], toke
 # Here starts the use on the engine object directly
 #
 
+def results_to_model(results, model: Type[BaseModel]) -> list[BaseModel]:
+    """Converts the results to a list of models"""
+
+    keys = list(results.keys())
+    return [
+        model(**{keys[i]: result[i] for i, v in enumerate(result)}) for result in results.fetchall()
+    ]
+
+
 class SQLResponse:
     def __init__(self, columns, results):
         self.columns = list(columns)
@@ -205,7 +215,7 @@ async def select_sources_sub_table(
         # Extract filters from the query parameters
         query_parser = QueryParser(columns=selected_columns, query_params=query_params)
         if query_parser.get_group_by_column() is not None:
-            selected_columns = query_parser.get_group_by_select_columns()
+            selected_columns = query_parser.get_select_columns()
 
         stmt = (
             select(*selected_columns)

api/models/object.py

@@ -0,0 +1,26 @@
+import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+from api.schemas import SchemeEnum
+
+
+class Object(BaseModel):
+    scheme: SchemeEnum
+    host: str
+    bucket: str
+    key: str
+    source: dict
+    mime_type: str
+    sha256_hash: str
+
+    class Config:
+        orm_mode = True
+
+
+class ResponseObject(Object):
+    id: int
+    created_on: datetime.datetime
+    updated_on: datetime.datetime
+    deleted_on: Optional[datetime.datetime] = None

api/models.py renamed to api/models/source.py

@@ -1,4 +1,3 @@
-import os
 from typing import Optional
 
 from geojson_pydantic import Feature, Polygon
@@ -72,4 +71,3 @@ class Config:
         orm_mode = True
 
 
-

api/query_parser.py

@@ -9,7 +9,7 @@
 
 import starlette.requests
 import logging
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Request
 
 
 from sqlalchemy.sql.expression import SQLColumnExpression
@@ -25,6 +25,12 @@ class ParserException(Exception):
     pass
 
 
+def get_filter_query_params(request: Request) -> list[tuple[str, str]]:
+    """Returns the query params that are not page or page_size"""
+
+    return [*filter(lambda x: x[0] not in ["page", "page_size"], request.query_params.items())]
+
+
 def cast_to_column_type(column: Column, value):
     try:
         return column.type.python_type(value)
@@ -91,11 +97,11 @@ def get_group_by_column(self):
 
         return group_by_columns[0]
 
-    def get_group_by_select_columns(self):
+    def get_select_columns(self) -> list[Column]:
         """Returns the group by expression which does a string aggregation of distinct values in the other columns"""
 
         if self.get_group_by_column() is None:
-            raise ParserException(f"Group by column must be set")
+            return [*self.columns.values()]
 
         columns = []
         for column in self.columns.values():

api/routes/object.py

@@ -0,0 +1,131 @@
+import datetime
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import insert, select, update, and_
+
+from api.database import (
+    get_async_session,
+    get_engine,
+    results_to_model
+)
+from api.routes.security import get_groups
+from api.models.object import Object, ResponseObject
+from api.schemas import Objects
+from api.query_parser import get_filter_query_params, QueryParser
+
+router = APIRouter(
+    prefix="/object",
+    tags=["file"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.get("/", response_model=list[ResponseObject])
+async def get_objects(page: int = 0, page_size: int = 50, filter_query_params=Depends(get_filter_query_params), groups: list[str] = Depends(get_groups)):
+    """Get all objects"""
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    query_parser = QueryParser(columns=Objects.__table__.c, query_params=filter_query_params)
+
+    async with async_session() as session:
+
+        # TODO: This flow should likely be refactored into a function, lets see it used once more before making the move
+        select_stmt = select(*query_parser.get_select_columns())\
+            .limit(page_size)\
+            .offset(page_size * page)\
+            .where(and_(Objects.deleted_on == None, query_parser.where_expressions()))
+
+        # Add grouping
+        if query_parser.get_group_by_column() is not None:
+            select_stmt = select_stmt.group_by(query_parser.get_group_by_column()).order_by(
+                query_parser.get_group_by_column()
+            )
+
+        if query_parser.get_order_by_columns() is not None and \
+                query_parser.get_group_by_column() is None:
+            select_stmt = select_stmt.order_by(*query_parser.get_order_by_columns())
+
+        results = await session.execute(select_stmt)
+
+        return results_to_model(results, ResponseObject)
+
+
+@router.get("/{id}", response_model=ResponseObject)
+async def get_object(id: int, groups: list[str] = Depends(get_groups)):
+    """Get a single object"""
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        select_stmt = select(Objects).where(and_(Objects.id == id, Objects.deleted_on == None))
+
+        result = await session.scalar(select_stmt)
+
+        if result is None:
+            raise HTTPException(status_code=404, detail=f"Object with id ({id}) not found")
+
+        response = ResponseObject(**result.__dict__)
+        return response
+
+
+@router.post("/", response_model=ResponseObject)
+async def create_file(object: Object, groups: list[str] = Depends(get_groups)):
+    """Create/Register a new object"""
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        insert_stmt = insert(Objects).values(**object.dict()).returning(Objects)
+        server_object = await session.scalar(insert_stmt)
+
+        response = ResponseObject(**server_object.__dict__)
+        await session.commit()
+        return response
+
+
+@router.patch("/{id}", response_model=ResponseObject)
+async def patch_object(id: int, object: Object, groups: list[str] = Depends(get_groups)) -> ResponseObject:
+    """Update a object"""
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        update_stmt = update(Objects)\
+            .where(Objects.id == id)\
+            .values(**object.dict())\
+            .returning(Objects)
+
+        server_object = await session.scalar(update_stmt)
+
+        response = ResponseObject(**server_object.__dict__)
+        await session.commit()
+        return response
+
+
+@router.delete("/{id}")
+async def delete_object(id: int, groups: list[str] = Depends(get_groups)) -> ResponseObject:
+    """Delete a object"""
+
+    engine = get_engine()
+    async_session = get_async_session(engine)
+
+    async with async_session() as session:
+
+        delete_stmt = update(Objects)\
+            .where(Objects.id == id)\
+            .values(deleted_on=datetime.datetime.utcnow())\
+            .returning(Objects)
+
+        server_object = await session.scalar(delete_stmt)
+
+        response = ResponseObject(**server_object.__dict__)
+        await session.commit()
+        return response

api/schemas.py

@@ -1,7 +1,8 @@
+import enum
 from typing import List
 import datetime
-from sqlalchemy import ForeignKey, func, DateTime
-from sqlalchemy.dialects.postgresql import VARCHAR, TEXT, INTEGER, ARRAY, BOOLEAN
+from sqlalchemy import ForeignKey, func, DateTime, Enum, UniqueConstraint
+from sqlalchemy.dialects.postgresql import VARCHAR, TEXT, INTEGER, ARRAY, BOOLEAN, JSON
 from sqlalchemy import String
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from geoalchemy2 import Geometry
@@ -83,3 +84,33 @@ class Token(Base):
     created_on: Mapped[datetime.datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
+
+
+class SchemeEnum(enum.Enum):
+    http = "http"
+    s3 = "s3"
+
+
+class Objects(Base):
+    __tablename__ = "objects"
+    __table_args__ = (
+        UniqueConstraint('scheme', 'host', 'bucket', 'key', name='unique_file'),
+        {'schema': 'macrostrat'}
+    )
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    scheme: Mapped[str] = mapped_column(Enum(SchemeEnum))
+    host: Mapped[str] = mapped_column(VARCHAR(255), nullable=False)
+    bucket: Mapped[str] = mapped_column(VARCHAR(255), nullable=False)
+    key: Mapped[str] = mapped_column(VARCHAR(255), nullable=False)
+    source: Mapped[dict] = mapped_column(JSON)
+    mime_type: Mapped[str] = mapped_column(VARCHAR(255))
+    sha256_hash: Mapped[str] = mapped_column(VARCHAR(255))
+    created_on: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    updated_on: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+    deleted_on: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )