added search-service files

Author: AasthathecoderX

services/search-service/app/core/cache.py

@@ -1,7 +1,17 @@
 import redis
-
 from app.core.config import settings
+# 1. Import the metrics you defined
+from app.core.metrics import cache_requests_total 
+
+redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True)
 
-# decode_responses=True means values come back as str instead of bytes --
-# convenient since we're storing JSON strings, not binary data.
-redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True)
+def get_from_cache(key: str):
+    data = redis_client.get(key)
+    if data:
+        # 2. Increment the metric on a HIT
+        cache_requests_total.labels(result="hit").inc()
+        return data
+    
+    # 3. Increment the metric on a MISS
+    cache_requests_total.labels(result="miss").inc()
+    return None

services/search-service/app/core/logging_config.py

@@ -0,0 +1,116 @@
+"""
+Structured logging setup, shared in shape across every FastAPI service in
+PriceLens (each service gets its own copy of this file -- there's no
+shared Python package across services in this project's architecture, so
+duplication here is the same deliberate tradeoff as the near-identical
+Dockerfiles from Phase 12: acceptable at this service count, would become
+a shared library at a larger scale).
+
+Two things this solves that plain `logging.basicConfig` doesn't:
+
+1. JSON output -- a real log aggregation tool (Loki, CloudWatch, ELK) can
+   parse and index individual fields (level, service, request_id) instead
+   of needing to regex-parse a plain text line. This is what makes a query
+   like "show me every ERROR from search-service in the last hour" fast
+   and exact rather than approximate.
+
+2. Request correlation ID -- generated once per incoming HTTP request and
+   attached to every log line produced while handling it, AND propagated
+   to any outgoing call to another service (via the X-Request-ID header).
+   This is what lets you grep one ID across auth-service, search-service,
+   and the Celery worker's logs and see the complete story of one user's
+   one request, rather than four separate unconnected timelines.
+"""
+import json
+import logging
+import time
+import uuid
+from contextvars import ContextVar
+
+from fastapi import Request
+from starlette.middleware.base import BaseHTTPMiddleware
+
+# ContextVar (not a plain global) is what makes this safe under concurrent
+# async requests -- each request gets its own isolated value even though
+# they're all running in the same process/event loop, unlike a plain
+# module-level variable which would leak between concurrent requests.
+request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")
+
+
+def get_current_request_id() -> str:
+    """
+    Used by outgoing HTTP clients (e.g. analytics-service's TrackingClient)
+    to read the current request's correlation ID and forward it as an
+    X-Request-ID header on calls to other services -- this is what makes
+    the ID actually correlate across service boundaries, not just within
+    one service's own logs.
+    """
+    return request_id_ctx.get()
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        payload = {
+            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "request_id": request_id_ctx.get(),
+        }
+        if record.exc_info:
+            payload["exception"] = self.formatException(record.exc_info)
+        return json.dumps(payload)
+
+
+def configure_logging(service_name: str) -> None:
+    handler = logging.StreamHandler()
+    handler.setFormatter(JsonFormatter())
+
+    root_logger = logging.getLogger()
+    root_logger.handlers = [handler]
+    root_logger.setLevel(logging.INFO)
+
+    # Tag every record with which service produced it -- useful once logs
+    # from multiple services are aggregated into one stream/dashboard.
+    old_factory = logging.getLogRecordFactory()
+
+    def record_factory(*args, **kwargs):
+        record = old_factory(*args, **kwargs)
+        record.service = service_name
+        return record
+
+    logging.setLogRecordFactory(record_factory)
+
+
+class RequestIdMiddleware(BaseHTTPMiddleware):
+    """
+    Reads X-Request-ID from an incoming request if present (meaning this
+    request originated from, or passed through, another PriceLens service
+    that already generated one), or creates a new one if this is the
+    entry point. Either way, sets it in the ContextVar so every log line
+    during this request includes it, and echoes it back in the response
+    header so a client (or another service) can correlate further.
+    """
+
+    async def dispatch(self, request: Request, call_next):
+        incoming_id = request.headers.get("X-Request-ID")
+        req_id = incoming_id or str(uuid.uuid4())
+        token = request_id_ctx.set(req_id)
+
+        start = time.monotonic()
+        try:
+            response = await call_next(request)
+        finally:
+            request_id_ctx.reset(token)
+
+        duration_ms = round((time.monotonic() - start) * 1000, 2)
+        response.headers["X-Request-ID"] = req_id
+
+        logging.getLogger("access").info(
+            "%s %s -> %s (%sms)",
+            request.method,
+            request.url.path,
+            response.status_code,
+            duration_ms,
+        )
+        return response

services/search-service/app/core/rate_limit.py

@@ -0,0 +1,17 @@
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+
+from app.core.config import settings
+
+# Reuses the SAME Redis instance already running since Phase 8 -- no new
+# infrastructure needed, just a different key namespace within it.
+# get_remote_address keys the limit by client IP; this is the standard
+# first line of defense against a single client hammering /search, which
+# matters specifically here because every cache miss (Phase 8) triggers a
+# REAL scrape (Phase 9) -- unrestrained request volume directly translates
+# into scrape volume, risking an Amazon/Flipkart IP ban (the exact risk
+# Phase 8's caching was originally introduced to reduce).
+limiter = Limiter(
+    key_func=get_remote_address,
+    storage_uri=settings.redis_url,
+)

services/search-service/app/main.py

@@ -2,17 +2,39 @@
 Application entrypoint.
 
 Deliberately thin: this file only wires things together (settings, routers).
-It must never contain business logic or database queries.
 """
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from prometheus_fastapi_instrumentator import Instrumentator
 
+from slowapi.errors import RateLimitExceeded
+from slowapi.middleware import SlowAPIMiddleware
+from slowapi import _rate_limit_exceeded_handler
+
 from app.api.v1 import health, search
 from app.core.config import settings
+from app.core.rate_limit import limiter
+
+app = FastAPI(
+    title=settings.app_name,
+    debug=settings.debug,
+)
+
+# -------------------------
+# Rate Limiting
+# -------------------------
+
+app.state.limiter = limiter
+app.add_exception_handler(
+    RateLimitExceeded,
+    _rate_limit_exceeded_handler,
+)
+app.add_middleware(SlowAPIMiddleware)
 
-app = FastAPI(title=settings.app_name, debug=settings.debug)
+# -------------------------
+# CORS
+# -------------------------
 
 app.add_middleware(
     CORSMiddleware,
@@ -22,12 +44,23 @@
     allow_headers=["*"],
 )
 
+# -------------------------
+# Routers
+# -------------------------
+
 app.include_router(health.router, prefix=settings.api_v1_prefix)
-app.include_router(search.router, prefix=settings.api_v1_prefix)
+app.include_router(search.router, prefix=settings.api_v1_prefix) 
 
+# -------------------------
+# Metrics
+# -------------------------
 
-@app.get("/")
-async def root() -> dict[str, str]:
-    return {"service": settings.app_name, "status": "running"}
-# Expose Prometheus metrics
 Instrumentator().instrument(app).expose(app)
+
+
+@app.get("/")
+async def root():
+    return {
+        "service": settings.app_name,
+        "status": "running",
+    }

services/search-service/app/tasks/scrape_tasks.py

@@ -0,0 +1,33 @@
+import logging
+from app.celery_app import celery_app
+from app.services.scraper_service import ScraperService
+# Import your centralized logging config to ensure worker logs match
+from app.core.logging_config import configure_logging, request_id_var
+
+# Initialize logging for the worker process
+configure_logging()
+logger = logging.getLogger(__name__)
+
+@celery_app.task(bind=True)
+def run_scrape_task(self, product_id: str, request_id: str = None):
+    """
+    Executes a product scrape task.
+    
+    :param product_id: The ID of the product to scrape.
+    :param request_id: Correlation ID propagated from the search-service.
+    """
+    # 1. Set the correlation ID for this worker task
+    # If no ID was provided (legacy task), fall back to the Celery Task ID
+    correlation_id = request_id or f"worker-{self.request.id}"
+    request_id_var.set(correlation_id)
+    
+    logger.info(f"Starting scrape task for product {product_id}")
+    
+    try:
+        scraper = ScraperService()
+        scraper.scrape(product_id)
+        logger.info(f"Successfully completed scrape task for {product_id}")
+    except Exception as e:
+        logger.error(f"Failed to scrape product {product_id}: {str(e)}")
+        # Re-raise to allow Celery's retry mechanism to handle it
+        raise e