feat: ♻️ Refactoring models to own queries against tables

foresight/stream_service/app.py

@@ -12,9 +12,8 @@
 import dotenv
 import requests
 
-from foresight.models.forex_data import ForexData
-from foresight.models.stream import Stream
-from foresight.utils.database import TimeScaleService
+from foresight.stream_service.models.forex_data import ForexData
+from foresight.stream_service.models.stream import Stream
 from foresight.utils.logger import generate_logger
 
 
@@ -59,7 +58,7 @@ def open_random_walk_stream(
             ask=round(initial_price + 0.0001, 5),
         )
 
-        TimeScaleService().insert_forex_data(record, table_name=table_name)
+        record.insert_forex_data(table_name=table_name)
 
         logger.info(record)
 
@@ -83,8 +82,8 @@ def process_stream_data(line: str, table_name: str = "forex_data"):
             logger.error(record.errorMessage)
         elif record.type == "PRICE" and record.tradeable:
 
-            TimeScaleService().insert_forex_data(
-                record.to_forex_data(),
+            forex_data = record.to_forex_data()
+            forex_data.insert_forex_data(
                 table_name=table_name,
             )
             logger.info(record)
@@ -135,28 +134,9 @@ def open_stream():
     execute_stream()
 
 
-def create_table(table_name: str = "forex_data") -> str:
-    """Create a table in the data store."""
-
-    # Execute SQL queries here
-    TimeScaleService().create_table(
-        query=f"""CREATE TABLE IF NOT EXISTS {table_name} (
-        instrument VARCHAR(10) NOT NULL,
-        time TIMESTAMPTZ NOT NULL,
-        bid FLOAT NOT NULL,
-        ask FLOAT NOT NULL,
-        PRIMARY KEY (instrument, time)
-    )""",
-        table_name=table_name,
-        column_name="time",
-    )
-
-    return table_name
-
-
 if __name__ == "__main__":
-    # Execute SQL queries here
-    create_table()
+    # Create the table in the data store if it does not exist.
+    ForexData.create_table()
 
     # Open a stream to the OANDA API and send the data to the data store.
     while True:

foresight/stream_service/models/forex_data.py

@@ -0,0 +1,73 @@
+"""Forex Data Model used in TimeScaleDB"""
+
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from foresight.utils.database import TimeScaleService
+
+
+class ForexData(BaseModel):
+    """TimescaleDB model for forex data.
+
+    Args:
+        instrument (str): The currency pair.
+        time (datetime): The time of the record.
+        bid (float): The bid price.
+        ask (float): The ask price.
+    """
+
+    instrument: str
+    time: datetime
+    bid: float
+    ask: float
+
+    @staticmethod
+    def create_table(table_name: str = "forex_data") -> str:
+        """Create a table in the data store if it does not exist.
+
+        Args:
+            table_name (str): The name of the table to create.
+
+        Returns:
+            str: The name of the table created.
+        """
+
+        # Execute SQL queries here
+        TimeScaleService().create_table(
+            query=f"""CREATE TABLE IF NOT EXISTS {table_name} (
+            instrument VARCHAR(10) NOT NULL,
+            time TIMESTAMPTZ NOT NULL,
+            bid FLOAT NOT NULL,
+            ask FLOAT NOT NULL,
+            PRIMARY KEY (instrument, time)
+        )""",
+            table_name=table_name,
+            column_name="time",
+        )
+
+        return table_name
+
+    def insert_forex_data(self, table_name: str = "forex_data"):
+        """Insert forex data into the database."""
+        TimeScaleService().execute(
+            query=f"""INSERT INTO {table_name} (instrument, time, bid, ask)
+            VALUES (%s, %s, %s, %s)""",
+            params=(
+                self.instrument,
+                self.time,
+                self.bid,
+                self.ask,
+            ),
+        )
+
+    @staticmethod
+    def drop_table(table_name: str = "forex_data"):
+        """Drop a table in the data store.
+
+        Args:
+            table_name (str): The name of the table to drop.
+        """
+
+        # Execute SQL queries here
+        TimeScaleService().execute(query=f"DROP TABLE {table_name}")

foresight/models/stream.py → foresight/stream_service/models/stream.py

@@ -7,8 +7,8 @@
 
 from pydantic import BaseModel
 
-from foresight.models.forex_data import ForexData
-from foresight.models.pricing import Pricing
+from foresight.stream_service.models.forex_data import ForexData
+from foresight.stream_service.models.pricing import Pricing
 
 
 class Stream(BaseModel):

foresight/utils/database.py

@@ -7,8 +7,6 @@
 import psycopg2
 import psycopg2.extras
 
-from foresight.models.forex_data import ForexData
-
 
 dotenv.load_dotenv(".env")
 
@@ -73,7 +71,9 @@ def create_table(self, query, table_name=None, column_name=None):
                         except psycopg2.DatabaseError:
                             logger.info("Already created the hyper table. Skipping.")
             except Exception as table_create_exception:
-                raise Exception(f"Failed to create table: {table_create_exception}")
+                raise ValueError(
+                    f"Failed to create table: {table_create_exception}",
+                ) from table_create_exception
 
         else:
             raise Exception("Database connection not established.")
@@ -98,19 +98,6 @@ def close(self):
             self.connection.close()
             self.connection = None
 
-    def insert_forex_data(self, forex_data: ForexData, table_name: str = "forex_data"):
-        """Insert forex data into the database."""
-        self.execute(
-            query=f"""INSERT INTO {table_name} (instrument, time, bid, ask)
-            VALUES (%s, %s, %s, %s)""",
-            params=(
-                forex_data.instrument,
-                forex_data.time,
-                forex_data.bid,
-                forex_data.ask,
-            ),
-        )
-
 
 # Example usage:
 # # Execute a sample query against native tables.

foresight/window_service/app.py

@@ -18,6 +18,9 @@
 logger = logging.getLogger(__name__)
 
 
+timeMap: dict = {"S": "second", "M": "minute", "H": "hour", "D": "day"}
+
+
 def fetch_data(instrument: str = "EUR_USD", timescale: str = "M") -> pd.DataFrame:
     """
     Fetch all data from the database and return a DataFrame.
@@ -30,42 +33,26 @@ def fetch_data(instrument: str = "EUR_USD", timescale: str = "M") -> pd.DataFram
         dict: The data from the database
     """
     try:
-        # Fetch all data from the database based on the parameters
-        if timescale == "M":
-            df = pd.DataFrame(
-                TimeScaleService().execute(
-                    query=f"""
-                        SELECT TO_CHAR(date_trunc('minute', time), 'YYYY-MM-DD HH24:MI:SS') as time,
-                        AVG(ask) as ask, AVG(bid) as bid
-                        FROM forex_data
-                        WHERE instrument = '{instrument}'
-                        AND time >= NOW() - INTERVAL '60 minute'
-                        GROUP BY time
-                        ORDER BY time ASC
-                    """,
-                ),
-            )
-
-            return (
-                df.groupby(df["time"]).agg({"ask": "mean", "bid": "mean"}).reset_index()
-            )
-        elif timescale == "T":
-            return pd.DataFrame(
-                TimeScaleService().execute(
-                    query=f"""
-                        SELECT TO_CHAR(time, 'YYYY-MM-DD HH24:MI:SS') as time,
-                        ask, bid
-                        FROM forex_data
-                        WHERE instrument = '{instrument}'
-                        AND time >= NOW() - INTERVAL '60 minute'
-                        ORDER BY time ASC
-                    """,
-                ),
-            )
-        else:
-            raise Exception(f"Invalid timescale: {timescale}")
-    except Exception as fetch_exception:
-        logger.error(f"Error: {fetch_exception}")
+        df = pd.DataFrame(
+            TimeScaleService().execute(
+                query=f"""
+                    SELECT
+                    TO_CHAR(
+                        date_trunc('{timeMap[timescale.lower()]}', time), 'YYYY-MM-DD HH24:MI:SS'
+                    ) as time,
+                    AVG(ask) as ask, AVG(bid) as bid
+                    FROM forex_data
+                    WHERE instrument = '{instrument}'
+                    AND time >= NOW() - INTERVAL '60 minute'
+                    GROUP BY time
+                    ORDER BY time ASC
+                """,
+            ),
+        )
+
+        return df.groupby(df["time"]).agg({"ask": "mean", "bid": "mean"}).reset_index()
+    except Exception as fetch_exception:  # pylint: disable=broad-except
+        logger.error("Error fetching data: %s", fetch_exception)
 
 
 if __name__ == "__main__":

tests/stream_service/conftest.py

@@ -0,0 +1,41 @@
+"""Test configuration for Foresight."""
+
+import pytest
+
+from foresight.stream_service.models.forex_data import ForexData
+from foresight.utils.database import TimeScaleService
+
+
+@pytest.fixture()
+def create_forex_data_table():
+    """Create a table in the TimescaleDB."""
+    table_name = ForexData.create_table()
+    yield table_name
+    ForexData.drop_table()
+
+
+@pytest.fixture()
+def insert_forex_data():
+    """Insert data into the TimescaleDB."""
+    # ARRANGE
+    table_name = create_forex_data_table
+    data = ForexData(
+        instrument="EUR_USD",
+        time="2021-01-01T00:00:00",
+        bid=1.0,
+        ask=1.0001,
+    )
+
+    # ACT
+    data.insert_forex_data()
+
+    # ASSERT
+    records = TimeScaleService().execute(
+        query=f"SELECT * FROM {table_name}",
+    )
+
+    assert len(records) == 1  # 1 record
+    assert records[0]["instrument"] == "EUR_USD"
+    assert records[0]["bid"] == 1.0
+    assert records[0]["ask"] == 1.0001
+    assert records[0]["time"] == "2021-01-01 00:00:00+00"