Added db persistence to event ingestion

Author: nir1r7

backend/app/api/routes/events.py

@@ -1,24 +1,38 @@
 from fastapi import APIRouter, status
 from app.models.event import EventIn
+from app.db import database
 from app.core.logging import get_logger
+import json
 
 router = APIRouter(prefix="/events", tags=["events"])
 logger = get_logger("eventrelay.events")
 
 @router.post("", status_code=status.HTTP_202_ACCEPTED)
 async def ingest_event(event: EventIn):
-    logger.info(
-        "event_recieved",
-        extra={
-            "extra": {
-                "source": event.source,
-                "type": event.type,
-                "timestamp": event.timestamp.isoformat(),
-            }
-        },
-    )
+    query = """
+    INSERT INTO events(source, type, payload)
+    VALUES (:source, :type, :payload)
+    RETURNING id;
+    """
+    # Convert payload dict to JSON string for JSONB column
+    values = event.dict()
+    values['payload'] = json.dumps(values['payload'])
+    
+    event_id = await database.execute(query = query, values=values)
+
+    logger.info(f"Stored event id={event_id} source={event.source} type={event.type}")
 
     return {
         "status": "accepted",
-        "event_type": event.type,
-    }
+        "id": event_id,
+    }
+
+@router.get("")
+async def get_all_events():
+    query = "SELECT * FROM events"
+
+    events = await database.fetch_all(query = query)
+
+    logger.info(f"Retrieved all events")
+
+    return events

backend/app/core/config.py

@@ -0,0 +1,14 @@
+from dotenv import load_dotenv
+import os
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent.parent
+load_dotenv(BASE_DIR / ".env")
+
+DB_USER = os.getenv("POSTGRES_USER")
+DB_PASSWORD  = os.getenv("POSTGRES_PASSWORD")
+DB_NAME = os.getenv("POSTGRES_DB", "eventrelay")
+DB_HOST = os.getenv("POSTGRES_HOST", "LOCALHOST")
+DB_PORT = os.getenv("POSTGRES_PORT", 5432)
+
+DATABASE_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

backend/app/db.py

@@ -0,0 +1,10 @@
+from databases import Database
+from app.core.config import DATABASE_URL
+
+database = Database(DATABASE_URL)
+
+async def connect_db():
+    await database.connect()
+
+async def disconnect_db():
+    await database.disconnect()

backend/app/main.py

@@ -1,5 +1,7 @@
 from fastapi import FastAPI
 from app.api.routes import events
+from app.db import connect_db, disconnect_db
+
 
 app = FastAPI(title="EventRelay")
 
@@ -8,3 +10,11 @@
 @app.get("/health")
 def health():
     return {"status": "ok"}
+
+@app.on_event("startup")
+async def startup():
+    await connect_db()
+
+@app.on_event("shutdown")
+async def shutdown():
+    await disconnect_db()

backend/app/models/event.py

@@ -6,7 +6,4 @@
 class EventIn(BaseModel):
     source: str = Field(..., example="auth-service")
     type: str = Field(..., example="user.login")
-    payload: Dict = Field(..., example={"user_id": 123})
-    timestamp: Optional[datetime] = Field(
-        default_factory=datetime.utcnow
-    )
+    payload: Dict = Field(..., example={"user_id": 123})