refactor: modernize query service

docker-compose.yaml

@@ -134,30 +134,35 @@ services:
       RINCON_ENDPOINT: "http://rincon:10311"
       SKIP_AUTH_CHECK: "true"
 
-  # query:
-  #   image: gauchoracing/mp_query:latest
-  #   container_name: query
-  #   restart: unless-stopped
-  #   depends_on:
-  #     - rincon
-  #   ports:
-  #     - "7010:7010"
-  #   environment:
-  #     ENV: "DEV"
-  #     PORT: "7010"
-  #     SERVICE_ENDPOINT: "http://query:7010"
-  #     SERVICE_HEALTH_CHECK: "http://query:7010/query/ping"
-  #     DATABASE_HOST: ${DATABASE_HOST}
-  #     DATABASE_PORT: ${DATABASE_PORT}
-  #     DATABASE_NAME: ${DATABASE_NAME}
-  #     DATABASE_USER: ${DATABASE_USER}
-  #     DATABASE_PASSWORD: ${DATABASE_PASSWORD}
-  #     RINCON_USER: "admin"
-  #     RINCON_PASSWORD: "admin"
-  #     RINCON_ENDPOINT: "http://rincon:10311"
-  #     SENTINEL_URL: ${SENTINEL_URL}
-  #     SENTINEL_JWKS_URL: ${SENTINEL_JWKS_URL}
-  #     SENTINEL_CLIENT_ID: ${SENTINEL_CLIENT_ID}
+  query:
+    build:
+      context: .
+      dockerfile: query/Dockerfile.dev
+    container_name: query
+    restart: unless-stopped
+    depends_on:
+      - rincon
+    ports:
+      - "7010:7010"
+    volumes:
+      - ./query:/app/query
+    environment:
+      ENV: "DEV"
+      PORT: "7010"
+      SERVICE_ENDPOINT: "http://query:7010"
+      SERVICE_HEALTH_CHECK: "http://query:7010/query/ping"
+      DATABASE_HOST: ${DATABASE_HOST}
+      DATABASE_PORT: ${DATABASE_PORT}
+      DATABASE_NAME: ${DATABASE_NAME}
+      DATABASE_USER: ${DATABASE_USER}
+      DATABASE_PASSWORD: ${DATABASE_PASSWORD}
+      RINCON_USER: "admin"
+      RINCON_PASSWORD: "admin"
+      RINCON_ENDPOINT: "http://rincon:10311"
+      SENTINEL_URL: ${SENTINEL_URL}
+      SENTINEL_JWKS_URL: ${SENTINEL_JWKS_URL}
+      SENTINEL_CLIENT_ID: ${SENTINEL_CLIENT_ID}
+      SKIP_AUTH_CHECK: "true"
 
   nanomq:
     image: emqx/nanomq:latest

query/Dockerfile

@@ -1,26 +1,13 @@
-FROM python:3.11-buster AS builder
+FROM python:3.12-slim-bookworm
 
-RUN pip install poetry==2.0.0
-
-ENV POETRY_NO_INTERACTION=1 \
-    POETRY_VIRTUALENVS_IN_PROJECT=1 \
-    POETRY_VIRTUALENVS_CREATE=1 \
-    POETRY_CACHE_DIR=/tmp/poetry_cache
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 
 WORKDIR /app
 
-COPY pyproject.toml poetry.lock ./
-RUN touch README.md
-
-RUN --mount=type=cache,target=$POETRY_CACHE_DIR poetry install --without dev --no-root
-
-FROM python:3.11-slim-buster AS runtime
-
-ENV VIRTUAL_ENV=/app/.venv \
-    PATH="/app/.venv/bin:$PATH"
-
-COPY --from=builder ${VIRTUAL_ENV} ${VIRTUAL_ENV}
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev --no-install-project
 
 COPY query ./query
+RUN uv sync --frozen --no-dev
 
-ENTRYPOINT ["python", "-m", "query.main"]
+ENTRYPOINT ["uv", "run", "python", "-m", "query.main"]

query/Dockerfile.dev

@@ -0,0 +1,7 @@
+FROM python:3.12-slim-bookworm
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /app/query
+
+CMD ["uv", "run", "uvicorn", "query.main:create_app", "--host", "0.0.0.0", "--port", "7010", "--reload"]

query/Makefile

@@ -1,7 +1,7 @@
 .PHONY: install run test lint clean
 
 install:
-	poetry install
+	uv sync
 
 run:
 	chmod +x scripts/run.sh
@@ -12,9 +12,9 @@ test:
 	./scripts/test.sh
 
 lint:
-	poetry run black .
-	poetry run isort .
-	poetry run flake8 .
+	uv run black .
+	uv run isort .
+	uv run flake8 .
 
 clean:
 	rm -rf __pycache__

query/pyproject.toml

@@ -8,29 +8,33 @@ authors = [
     {name = "Madhav Viswesvaran", email = "madhav@ucsb.edu"},
 ]
 readme = "README.md"
-requires-python = ">=3.10,<4.0"
+requires-python = ">=3.12,<4.0"
 dependencies = [
-    "fastapi (>=0.115.10,<0.116.0)",
-    "uvicorn (>=0.34.0,<0.35.0)",
-    "dotenv (>=0.9.9,<0.10.0)",
-    "sqlalchemy (>=2.0.38,<3.0.0)",
-    "pymysql (>=1.1.1,<2.0.0)",
-    "numpy (>=2.2.3,<3.0.0)",
-    "pandas (>=2.2.3,<3.0.0)",
-    "loguru (>=0.7.3,<0.8.0)",
-    "pyarrow (>=20.0.0,<21.0.0)",
-    "requests (>=2.32.3,<3.0.0)",
-    "pyjwt[crypto] (>=2.11.0,<3.0.0)"
+    "fastapi>=0.115.10,<0.116.0",
+    "uvicorn>=0.34.0,<0.35.0",
+    "dotenv>=0.9.9,<0.10.0",
+    "sqlalchemy>=2.0.38,<3.0.0",
+    "psycopg2-binary>=2.9.10,<3.0.0",
+    "numpy>=2.2.3,<3.0.0",
+    "pandas>=2.2.3,<3.0.0",
+    "loguru>=0.7.3,<0.8.0",
+    "pyarrow>=20.0.0,<21.0.0",
+    "requests>=2.32.3,<3.0.0",
+    "pyjwt[crypto]>=2.11.0,<3.0.0",
+    "mapache-py>=3.0.1,<4.0.0",
+    "rincon>=1.0.0,<2.0.0",
+    "gr-ulid>=1.1.2,<2.0.0",
 ]
 
+[dependency-groups]
+dev = [
+    "pytest>=8.3.5",
+    "pytest-cov>=6.1.1",
+]
 
-[build-system]
-requires = ["poetry-core>=2.0.0,<3.0.0"]
-build-backend = "poetry.core.masonry.api"
-
-[tool.poetry.scripts]
+[project.scripts]
 query = "query.main:main"
-[tool.poetry.group.dev.dependencies]
-pytest = "^8.3.5"
-pytest-cov = "^6.1.1"
 
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

query/query/config/config.py

@@ -22,12 +22,12 @@ class Config:
     RINCON_ENDPOINT: str = os.getenv('RINCON_ENDPOINT')
 
     # Auth settings
+    SKIP_AUTH_CHECK: bool = os.getenv('SKIP_AUTH_CHECK', 'false').lower() == 'true'
     SENTINEL_URL: str = os.getenv('SENTINEL_URL')
     SENTINEL_JWKS_URL: str = os.getenv('SENTINEL_JWKS_URL')
     SENTINEL_CLIENT_ID: str = os.getenv('SENTINEL_CLIENT_ID')
 
     @staticmethod
     def get_database_url() -> str:
-        """Constructs the MySQL database URL from individual settings"""
-        return f"mysql+pymysql://{Config.DATABASE_USER}:{Config.DATABASE_PASSWORD}@{Config.DATABASE_HOST}:{Config.DATABASE_PORT}/{Config.DATABASE_NAME}"
+        return f"postgresql+psycopg2://{Config.DATABASE_USER}:{Config.DATABASE_PASSWORD}@{Config.DATABASE_HOST}:{Config.DATABASE_PORT}/{Config.DATABASE_NAME}"
 

query/query/main.py

@@ -1,22 +1,37 @@
-from fastapi import FastAPI  
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-import requests
+from loguru import logger
 import uvicorn
 from query.config.config import Config
 from query.database.connection import init_db
 from query.routes import ping, query, signal_definition, token
 from query.service.auth import AuthService
-from query.service.rincon import RinconService
-from query.service.trip import get_all_trips, get_trip_by_id
-from query.service.vehicle import get_all_vehicles, get_vehicle_by_id
+from query.service.rincon import init_rincon
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    init_db()
+    init_rincon()
+    if Config.SKIP_AUTH_CHECK:
+        logger.warning("SKIP_AUTH_CHECK is enabled, skipping Sentinel initialization")
+    else:
+        AuthService.configure(
+            jwks_url=Config.SENTINEL_JWKS_URL,
+            issuer="https://sso.gauchoracing.com",
+            audience=Config.SENTINEL_CLIENT_ID
+        )
+    yield
 
 def create_app():
     app = FastAPI(
         title="Gaucho Racing Query",
         description="API Documentation",
         version=Config.VERSION,
         docs_url="/query/docs",
-        redoc_url="/query/redoc"
+        redoc_url="/query/redoc",
+        lifespan=lifespan
     )
 
     app.add_middleware(
@@ -32,7 +47,7 @@ def create_app():
         prefix="/query",
         tags=["Ping"]
     )
-    
+
     app.include_router(
         query.router,
         prefix="/query",
@@ -50,20 +65,12 @@ def create_app():
         prefix="/query",
         tags=["Token"]
     )
-    
+
     return app
 
 def main():
-  init_db()
-  RinconService.register()
-  AuthService.configure(
-    jwks_url=Config.SENTINEL_JWKS_URL,
-    issuer="https://sso.gauchoracing.com",
-    audience=Config.SENTINEL_CLIENT_ID
-  )
-
-  app = create_app()
-  uvicorn.run(app, host="0.0.0.0", port=Config.PORT)
+    app = create_app()
+    uvicorn.run(app, host="0.0.0.0", port=Config.PORT)
 
 if __name__ == "__main__":
-  main()
+    main()

query/query/routes/query.py

@@ -3,16 +3,17 @@
 from typing import Annotated
 from loguru import logger
 from fastapi.responses import JSONResponse
-import pandas as pd
-from query.model.log import QueryLog
-from query.service.auth import AuthService
-from query.service.log import create_log
-from query.service.query import query_signals, merge_to_smallest, merge_to_largest
 import numpy as np
 import traceback
 
+from query.config.config import Config
+from query.model.log import QueryLog
+from query.service.auth import AuthService
+from query.service.log import create_log
+from query.service.query import query_signals, merge_signals
 from query.service.token import get_token_by_id, validate_token
 from query.service.trip import get_trip_by_id
+import pandas as pd
 
 router = APIRouter()
 
@@ -32,7 +33,9 @@ async def get_signals(
 ):
     user_id = None
     try:
-        if authorization and "Bearer " in authorization:
+        if Config.SKIP_AUTH_CHECK:
+            user_id = "mock-user"
+        elif authorization and "Bearer " in authorization:
             logger.info(f"Found bearer token: {authorization}")
             auth_token = authorization.split("Bearer ")[1]
             user_id = AuthService.get_user_id_from_token(auth_token)
@@ -54,17 +57,17 @@ async def get_signals(
                     "message": "you are not authorized to access this resource",
                 }
             )
-        
+
         logger.info(f"Successfully authenticated user: {user_id}")
-        
+
         if vehicle_id is None:
             return JSONResponse(
                 status_code=400,
                 content={
                     "message": "vehicle_id is required",
                 }
             )
-        
+
         if signals is None or len(signals.split(",")) == 0 or any(not s.strip() for s in signals.split(",")):
             return JSONResponse(
                 status_code=400,
@@ -109,30 +112,18 @@ async def get_signals(
                     return JSONResponse(
                         status_code=400,
                         content={
-                            "message": "invalid end timestamp format", 
+                            "message": "invalid end timestamp format",
                         }
                     )
-            
+
         start_time = datetime.now()
         dfs = query_signals(vehicle_id=vehicle_id, signals=signals.split(","), start=start, end=end)
 
-        if merge == 'smallest':
-            merged_df, metadata = merge_to_smallest(*dfs, tolerance=tolerance, fill=fill)
-        elif merge == 'largest':
-            merged_df, metadata = merge_to_largest(*dfs, tolerance=tolerance, fill=fill)
-        else:
-            return JSONResponse(
-                status_code=400,
-                content={
-                    "message": "invalid merge strategy", 
-                }
-            )
-
-        # Convert timestamps to ISO format strings and handle special float values
+        merged_df, metadata = merge_signals(*dfs, strategy=merge, tolerance=tolerance, fill=fill)
+
         df_dict = merged_df.copy()
-        df_dict['produced_at'] = df_dict['produced_at'].dt.tz_localize('UTC').dt.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
-
-        # Replace inf/-inf and NaN values with None
+        df_dict['produced_at'] = df_dict['produced_at'].dt.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
+
         df_dict = df_dict.replace([np.inf, -np.inf], None)
         df_dict = df_dict.replace({np.nan: None})
 
@@ -170,7 +161,7 @@ async def get_signals(
             parquet_data = df_dict.to_parquet()
             return Response(
                 content=parquet_data,
-                media_type="application/octet-stream", 
+                media_type="application/octet-stream",
                 headers={
                     "Content-Disposition": "attachment; filename=export.parquet"
                 }
@@ -198,4 +189,4 @@ async def get_signals(
             content={
                 "message": str(e),
             }
-        )
+        )