Nuke schemas.py

src/main.py

@@ -2,7 +2,7 @@
 
 from src.auth import verify_jwt
 from src.redis import set_redis_from_tx
-from src.schemas import AnalyticsTxn
+from src.models import AnalyticsTxn
 
 app = FastAPI()
 

src/models.py

@@ -1,5 +1,7 @@
 import hashlib
 from enum import Enum
+from datetime import datetime
+from typing import Optional
 
 from fastapi.encoders import jsonable_encoder
 from pydantic import BaseModel, ConfigDict, json
@@ -45,3 +47,37 @@ def __str__(self):
 class RedisEvent(CustomModel):
     key: bytes | str
     value: bytes | str
+
+class AnalyticsTxn(CustomModel):
+    product: Product
+    pennkey: Optional[str] = None
+    timestamp: datetime
+    data: list[RedisEvent]
+
+    # init with JSON data
+    def __init__(self, **data):
+        super().__init__(**data)
+        self.timestamp = datetime.fromtimestamp(data["timestamp"])
+        self.data = [RedisEvent(**event) for event in data["data"]]
+        self.product = Product(data["product"])
+        self.pennkey = data.get("pennkey")
+
+    def get_redis_key(self):
+        return f"{self.product}.{self.hash_as_key()}"
+
+    def build_redis_data(self) -> list[RedisEvent]:
+        return [
+            RedisEvent(
+                key=f"{self.get_redis_key()}.{event.hash_as_key()}",
+                value=json.dumps(
+                    {
+                        "product": str(self.product),
+                        "pennkey": self.pennkey,
+                        "timestamp": self.timestamp.timestamp(),
+                        "datapoint": event.key,
+                        "value": event.value,
+                    }
+                ),
+            )
+            for event in self.data
+        ]

src/redis.py

@@ -3,7 +3,7 @@
 from redis.asyncio import Redis
 from src.config import settings
 from src.models import RedisEvent
-from src.schemas import AnalyticsTxn
+from src.models import AnalyticsTxn
 
 redis_client: Redis = Redis.from_url(str(settings.REDIS_URL))
 

tests/test_redis.py

@@ -1,7 +1,7 @@
 import pytest
 
 from src.redis import get_by_key, set_redis_keys
-from src.schemas import RedisEvent
+from src.models import RedisEvent
 
 
 @pytest.mark.asyncio