Edit

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.10
+
+WORKDIR /app
+
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,68 @@
+import pathway as pw
+import threading
+from fastapi import FastAPI, HTTPException
+import uvicorn
+import os
+from exporter import export_loop
+
+class ProductSchema(pw.Schema):
+    _id: str
+    name: str
+    price: float
+    description: str
+    category: str
+    stock: int
+    imageUrl: str
+    isNew: bool
+    isActive: bool
+    createdBy: str
+    ratings: dict
+    createdAt: str
+    updatedAt: str
+    promoCopy: str
+
+cache = {}
+cache_lock = threading.Lock()
+app = FastAPI()
+
+def run_pipeline():
+    products = pw.io.jsonlines.read(
+        "mongo_exports/products.jsonl",
+        schema=ProductSchema,
+        mode="streaming"
+    )
+    transformed = products.select(
+        _id=pw.this._id,
+        name=pw.this.name,
+        price=pw.this.price,
+        category=pw.this.category,
+        promo=pw.this.promoCopy
+    ).with_id_from(pw.this._id)
+
+    def update_cache(key, row, time, is_addition):
+        with cache_lock:
+            if is_addition:
+                cache[str(key)] = row
+            else:
+                cache.pop(str(key), None)
+
+    pw.io.subscribe(transformed, on_change=update_cache)
+    pw.run()
+
+@app.get("/products")
+def get_all_products():
+    with cache_lock:
+        return list(cache.values())
+
+@app.get("/products/{product_id}")
+def get_product(product_id: str):
+    with cache_lock:
+        product = cache.get(product_id)
+        if not product:
+            raise HTTPException(status_code=404, detail="Product not found")
+        return product
+
+if __name__ == "__main__":
+    threading.Thread(target=export_loop, daemon=True).start()
+    threading.Thread(target=run_pipeline, daemon=True).start()
+    uvicorn.run(app, host="0.0.0.0", port=7860)

exporter.py

@@ -0,0 +1,31 @@
+from pymongo import MongoClient
+from bson import ObjectId
+from datetime import datetime
+import json
+import time
+import os
+
+client = MongoClient("mongodb+srv://rivalcoder01:PurmpaRXBSThyuQA@cluster0.f7zziod.mongodb.net/?retryWrites=true&w=majority")
+collection = client["test"]["products"]
+
+EXPORT_DIR = "mongo_exports"
+os.makedirs(EXPORT_DIR, exist_ok=True)
+
+def clean(doc):
+    for k, v in doc.items():
+        if isinstance(v, ObjectId):
+            doc[k] = str(v)
+        elif isinstance(v, datetime):
+            doc[k] = v.isoformat()
+    return doc
+
+def export_loop():
+    while True:
+        docs = list(collection.find())
+        fname = f"{EXPORT_DIR}/products.jsonl"
+        with open(fname, "w") as f:
+            for doc in docs:
+                doc = clean(doc)
+                f.write(json.dumps(doc) + "\n")
+        print("[Mongo Exporter] Wrote updated products.jsonl")
+        time.sleep(5)

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+pymongo
+pathway
+bson