Merge pull request #28 from ATNoG/feature/query_raw_data

Feature/query raw data

Author: alxmra

main.py

@@ -5,7 +5,9 @@
 from fastapi import FastAPI
 
 from src.routers.v1 import v1_router
-from src.routers.v1.latency import ClickHouse
+from src.routers.v1.latency_router import ClickHouse
+from src.routers.v1.raw_router import Influx
+
 from src.sink import KafkaSinkManager
 
 KAFKA_HOST = os.getenv("KAFKA_HOST", "localhost")
@@ -15,6 +17,8 @@
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+
+    Influx.service.connect()
     ClickHouse.service.connect()
 
     sink_manager = KafkaSinkManager(KAFKA_HOST, KAFKA_PORT)

src/models/raw.py

@@ -5,6 +5,8 @@
 
 RAW:str = "raw"
 
+
+
 class Raw(BaseModel):
     # Temporal
     timestamp: datetime = Field(
@@ -49,3 +51,7 @@ def to_point(self) -> Point:
                 else:
                     point.field(key, value)
         return point
+
+class RawResponse(BaseModel):
+    data: list[Raw]
+    has_next: bool

src/routers/v1/__init__.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
-from src.routers.v1 import latency
+from src.routers.v1.latency_router import router as latencyR
+from src.routers.v1.raw_router import router as rawR
 
 v1_router = APIRouter()
-
-v1_router.include_router(latency.router, prefix="/processed", tags=["v1", "latency"])
+v1_router.include_router(latencyR, prefix="/processed", tags=["v1", "latency"])
+v1_router.include_router(rawR, prefix="/raw", tags=["v1", "latency"])

src/routers/v1/latency.py src/routers/v1/latency_router.pysrc/routers/v1/latency.py renamed to src/routers/v1/latency_router.py

@@ -0,0 +1,47 @@
+from fastapi import APIRouter, HTTPException, Query
+from src.configs.influx_conf import InfluxConf
+from src.services.influx import InfluxService
+from src.models.raw import RawResponse
+from datetime import datetime
+
+
+router = APIRouter()
+
+class Influx():
+    conf = InfluxConf()
+    service = InfluxService()
+
+
+@router.get("/", response_model=RawResponse)
+def get_raw_data(
+    start_time: int = Query(..., description="Start time (Unix timestamp in seconds)"),
+    end_time: int = Query(..., description="End time (Unix timestamp in seconds)"),
+    cell_index: int = Query(..., description="Cell index (required)"),
+
+    batch_number: int = Query(1, ge=1, description="Batch number"),
+):
+    """
+    Query raw data entries between a given start and end time for a specific cell.
+
+    This endpoint returns raw entries directly from the database,
+    limited to a maximum of 50 per request.
+    """
+    try:
+        query_params = {
+            "start_time": start_time,
+            "end_time": end_time,
+            "cell_index": cell_index,
+            "batch_number": batch_number,
+        }
+
+        results,has_next = Influx.service.query_raw_data(**query_params)
+
+        for row in results:
+            for k, v in row.items():
+                if isinstance(v, datetime):
+                    row[k] = v.isoformat()
+
+        return {"data": results, "has_next": has_next}
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error querying raw data: {str(e)}")

src/routers/v1/raw_router.py

@@ -4,7 +4,7 @@
 from src.configs.clickhouse_conf import ClickhouseConf
 from src.models.processed_latency import ProcessedLatency
 from src.services.db_service import DBService
-from src.services.query import QueryCH
+from src.services.clickhouse_query import QueryCH
 
 
 def transform_processor_output(data: dict) -> dict:
@@ -83,6 +83,7 @@ def __init__(self) -> None:
         self.conf =   ClickhouseConf()
         self.client:  Client = None
 
+
     def connect(self):
         self.client = clickhouse_connect.get_client(
             host     = self.conf.host,

src/services/clickhouse.py

@@ -4,13 +4,11 @@
 from src.configs.influx_conf import InfluxConf
 from src.models.raw import Raw
 from src.services.db_service import DBService
+from src.services.influx_query import QueryIF
 
 class InfluxService(DBService):
     def __init__(self) -> None:
         self.conf=      InfluxConf()
-        self.client:    InfluxDBClient
-        self.write_api: WriteApi
-        self.query_api: QueryApi
 
     def connect(self):
         """Init client"""
@@ -28,6 +26,30 @@ def get_data(self, batch_number:int = 1 ,batch_size:int = 50) -> list[Raw]:
         #TODO: add query
         return []
 
+    def query_raw_data(self, start_time: int, end_time: int, cell_index: int, batch_number: int):
+        _LIMIT = 50
+        offset = (batch_number-1) * _LIMIT
+
+        query = QueryIF.between.format(
+            bucket=self.conf.bucket,
+            start_time=start_time,
+            end_time=end_time,
+            cell_index=cell_index,
+            limit=_LIMIT+1,
+            offset=offset
+        )
+
+        result = self.query_api.query(query)
+
+        # Convert FluxTable to list of dicts
+        rows = []
+        for table in result:
+            for record in table.records:
+                rows.append(record.values)
+
+        return rows, len(rows)>_LIMIT
+
+
     def write_data(self, data: dict) -> None :
         """Write a single record"""
         raw = Raw(**data)

src/services/query.py src/services/clickhouse_query.pysrc/services/query.py renamed to src/services/clickhouse_query.py

@@ -0,0 +1,7 @@
+class QueryIF:
+    between = """
+    from(bucket: "{bucket}")
+      |> range(start: {start_time}, stop: {end_time})
+      |> filter(fn: (r) => r["cell_index"] == "{cell_index}")
+      |> limit(n: {limit}, offset: {offset})
+    """

src/services/influx.py

@@ -13,7 +13,7 @@
 @pytest.fixture
 def mock_clickhouse_service():
     """Mock ClickHouseService for testing."""
-    with patch('src.routers.v1.latency.ClickHouse') as mock:
+    with patch('src.routers.v1.latency_router.ClickHouse') as mock:
         service_mock = MagicMock()
         mock.service = service_mock
         yield service_mock