Refactor magstats to fast api

.github/workflows/tests.yml

@@ -28,3 +28,10 @@ jobs:
         test-folder: 'tests/integration'
         base-folder: 'astroobject'
         codecov-flags: ""
+  magstats_tests:
+    uses: ./.github/workflows/test_template.yaml
+    with:
+        python-version: '3.11'
+        sources-folder: 'magstats'
+        test-folder: 'tests'
+        base-folder: 'magstats'

magstats/Dockerfile

@@ -0,0 +1,30 @@
+FROM python:3.11 as python-base
+LABEL org.opencontainers.image.authors="ALeRCE"
+ENV PYTHONDONTWRITEBYTECODE=1 \
+  PYTHONUNBUFFERED=1 \
+  PYTHONFAULTHANDLER=1 \
+  PIP_NO_CACHE_DIR=off \
+  PIP_DISABLE_PIP_VERSION_CHECK=on \
+  PIP_DEFAULT_TIMEOUT=100 \
+  POETRY_VIRTUALENVS_IN_PROJECT=true \
+  POETRY_NO_INTERACTION=1
+
+
+FROM python-base as builder
+RUN pip install poetry
+WORKDIR /app
+COPY ./poetry.lock ./pyproject.toml /app
+RUN poetry install --no-root --without=dev
+
+
+FROM python:3.11-slim as production
+RUN pip install poetry
+COPY --from=builder /app /app
+WORKDIR /app
+COPY ./README.md /app/README.md
+COPY ./entrypoint.sh /app/
+COPY ./core /app/core
+COPY ./api /app/api
+COPY ./database /app/database
+RUN poetry install --only-root
+CMD ["bash", "entrypoint.sh"]

magstats/api/api.py

@@ -0,0 +1,23 @@
+import os
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi import FastAPI
+from .routes import router
+from prometheus_fastapi_instrumentator import Instrumentator
+
+app = FastAPI()
+instrumentator = Instrumentator().instrument(app)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(router)
+
+
+@app.on_event("startup")
+async def _startup():
+    instrumentator.expose(app)

magstats/api/result_handler.py

@@ -0,0 +1,27 @@
+from fastapi import HTTPException
+from core.exceptions import (
+    DatabaseError,
+    OidError,
+)
+import logging
+
+
+def handle_success(result):
+    return result
+
+
+def handle_error(err: Exception):
+    if isinstance(err, DatabaseError):
+        _handle_server_error(err)
+    if isinstance(err, OidError):
+        _handle_client_error(err)
+
+
+def _handle_server_error(err: Exception):
+    logging.error(err)
+    raise HTTPException(status_code=500, detail=str(err))
+
+
+def _handle_client_error(err: Exception):
+    print(str(err))
+    raise HTTPException(status_code=400, detail=str(err))

magstats/api/routes.py

@@ -0,0 +1,24 @@
+from fastapi import Request, APIRouter
+from core.service import get_magstats
+from .result_handler import handle_success, handle_error
+from database.sql import session
+
+router = APIRouter()
+
+
+@router.get("/")
+def root():
+    return "this is the magstats module"
+
+
+@router.get("/magstats/{oid}")
+def magstats(
+    request: Request,
+    oid: str,
+):
+    return get_magstats(
+        oid=oid,
+        session_factory=session,
+        handle_success=handle_success,
+        handle_error=handle_error,
+    )

magstats/core/exceptions.py

@@ -0,0 +1,22 @@
+class WrapperException(BaseException):
+    def __init__(self, original_e, subcode=None):
+        super().__init__()
+        self.original_exception = original_e
+        self.subcode = subcode
+
+    def __str__(self) -> str:
+        return self.original_exception.__str__()
+
+
+class DatabaseError(WrapperException):
+    def __init__(self, original_e, subcode=None):
+        super().__init__(original_e, subcode)
+
+
+class OidError(BaseException):
+    def __init__(self, oid) -> None:
+        super().__init__()
+        self.oid = oid
+
+    def __str__(self) -> str:
+        return f"Can't retrieve magstats oid not recognized {self.oid}"

magstats/core/models.py

@@ -0,0 +1,33 @@
+from pydantic import BaseModel, Field
+
+
+class MagstatsModel(BaseModel):
+    fid: int = Field(description="Filter ID (1=g; 2=r, 3=i)")
+    stellar: bool = Field(
+        description="Whether the object appears to be unresolved in the given band"
+    )
+    corrected: bool = Field(
+        description="Whether the corrected photometry should be used"
+    )
+    ndet: int = Field(description="Number of detections in the given band")
+    ndubious: int = Field(description="Number of dubious corrections")
+    magmean: float = Field(description="The mean magnitude for the given fid")
+    magmedian: float = Field(
+        description="The median magnitude for the given fid"
+    )
+    magmax: float = Field(description="The max magnitude for the given fid")
+    magmin: float = Field(description="The min magnitude for the given fid")
+    magsigma: float = Field(
+        description="Magnitude standard deviation for the given fid"
+    )
+    maglast: float = Field(description="The last magnitude for the given fid")
+    magfirst: float = Field(
+        description="The first magnitude for the given fid"
+    )
+    firstmjd: float = Field(
+        description="The time of the first detection in the given fid"
+    )
+    lastmjd: float = Field(
+        description="The time of the last detection in the given fid"
+    )
+    step_id_corr: str = Field(description="Correction step pipeline version")

magstats/core/service.py

@@ -0,0 +1,51 @@
+from contextlib import AbstractContextManager
+from typing import Callable
+from .models import MagstatsModel
+from pymongo.database import Database
+from sqlalchemy.orm import Session
+from .exceptions import DatabaseError, OidError
+from db_plugins.db.sql import models
+from sqlalchemy import select, text
+
+
+def default_handle_success(result):
+    return result
+
+
+def default_handle_error(error):
+    raise error
+
+
+def get_magstats(
+    oid: str,
+    session_factory: Callable[..., AbstractContextManager[Session]] = None,
+    handle_success: Callable[..., dict] = default_handle_success,
+    handle_error: Callable[Exception, None] = default_handle_error,
+):
+    result = _get_magstats_sql(session_factory, oid, handle_error)
+    if len(result) == 0:
+        handle_error(OidError(oid))
+    else:
+        print(result)
+        return handle_success(result)
+
+
+def _get_magstats_sql(
+    session_factory: Callable[..., AbstractContextManager[Session]],
+    oid: str,
+    handle_error: Callable[Exception, None] = default_handle_error,
+):
+    try:
+        with session_factory() as session:
+            stmt = select(models.MagStats, text("'ztf'")).filter(
+                models.MagStats.oid == oid
+            )
+            result = session.execute(stmt)
+            result = [
+                MagstatsModel(**ob[0].__dict__, tid=ob[1])
+                for ob in result.all()
+            ]
+            return result
+
+    except Exception as e:
+        return handle_error(DatabaseError(e))

magstats/database/sql.py

@@ -0,0 +1,36 @@
+from contextlib import contextmanager, AbstractContextManager
+import logging
+import os
+from typing import Callable
+from sqlalchemy.orm import scoped_session, sessionmaker, Session
+from sqlalchemy.engine import Engine, create_engine
+
+
+logger = logging.getLogger(__name__)
+
+
+user = os.getenv("PSQL_USER")
+pwd = os.getenv("PSQL_PASSWORD")
+host = os.getenv("PSQL_HOST")
+port = os.getenv("PSQL_PORT")
+db = os.getenv("PSQL_DATABASE")
+db_url = f"postgresql://{user}:{pwd}@{host}:{port}/{db}"
+
+engine: Engine = create_engine(db_url, echo=False)
+
+
+@contextmanager
+def session() -> Callable[..., AbstractContextManager[Session]]:
+    session_factory = scoped_session(
+        sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    )
+    session: Session = session_factory()
+    try:
+        yield session
+    except Exception:
+        logger.debug("Connecting databases")
+        logger.exception("Session rollback because of exception")
+        session.rollback()
+        raise
+    finally:
+        session.close()

magstats/entrypoint.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env sh
+
+poetry run uvicorn --port $PORT api.api:app