refactor(ans): Restructure ANS microservice architecture and configuration

- Reorganize project structure with new directories for models, repositories, and services
- Add comprehensive README.md with service documentation and setup instructions
- Create .env.example with detailed environment configuration
- Update .gitignore to include comprehensive Python and development exclusions
- Modify app.py to include logging setup and lifespan context manager
- Remove deprecated service and route files
- Add client modules for inter-service communication
- Implement utility modules for database and health checks
- Enhance project configuration and dependency management

.env.example

@@ -0,0 +1,22 @@
+# ANS Environment Variables
+
+# Database Configuration
+DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/ans_db
+REDIS_URL=redis://localhost:6379/0
+MONGODB_URL=mongodb://localhost:27017/ans_db
+
+# Service URLs for Inter-Service Communication
+MPMS_BASE_URL=http://localhost:8001
+RMS_BASE_URL=http://localhost:8002
+TMS_BASE_URL=http://localhost:8003
+CRM_BASE_URL=http://localhost:8004
+
+# Service Configuration
+SERVICE_NAME=ans
+SERVICE_PORT=8005
+LOG_LEVEL=INFO
+
+# JWT Configuration
+JWT_SECRET_KEY=your-secret-key
+JWT_ALGORITHM=HS256
+JWT_EXPIRATION_MINUTES=1440

.gitignore

@@ -0,0 +1,145 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db

README.md

@@ -1,10 +1,85 @@
----
-title: Insightfy Bloom Ms Ans
-emoji: 🏃
-colorFrom: red
-colorTo: pink
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Analytics and Notification Service (ANS)
+
+The Analytics and Notification Service (ANS) is a microservice that provides business intelligence, analytics, and notification capabilities for the Insightfy Bloom platform.
+
+## Features
+
+- **Analytics Dashboard**: Real-time business metrics and KPIs
+- **Reporting**: Generate various business reports
+- **Data Aggregation**: Collect and process data from other microservices
+- **Notifications**: Send alerts and notifications based on business rules
+- **Performance Monitoring**: Track business performance indicators
+
+## Architecture
+
+ANS integrates with other microservices:
+- **MPMS**: Product and merchant data
+- **RMS**: Resource and employee data  
+- **TMS**: Transaction and sales data
+- **CRM**: Customer relationship data
+
+## Database Connections
+
+- **PostgreSQL**: Primary database for analytics data
+- **MongoDB**: Document storage for flexible analytics
+- **Redis**: Caching and session management
+
+## API Endpoints
+
+### Health Checks
+- `GET /health` - Service health status
+- `GET /ready` - Readiness check
+- `GET /live` - Liveness check
+
+### Analytics
+- `GET /api/v1/analytics/dashboard` - Dashboard data
+- `GET /api/v1/analytics/reports` - Generate reports
+
+## Environment Variables
+
+Copy `.env.example` to `.env` and configure:
+
+```bash
+# Database Configuration
+DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/ans_db
+REDIS_URL=redis://localhost:6379/0
+MONGODB_URL=mongodb://localhost:27017/ans_db
+
+# Service URLs
+MPMS_BASE_URL=http://localhost:8001
+RMS_BASE_URL=http://localhost:8002
+TMS_BASE_URL=http://localhost:8003
+CRM_BASE_URL=http://localhost:8004
+
+# Service Configuration
+SERVICE_NAME=ans
+SERVICE_PORT=8005
+LOG_LEVEL=INFO
+
+# JWT Configuration
+JWT_SECRET_KEY=your-secret-key
+JWT_ALGORITHM=HS256
+```
+
+## Running the Service
+
+```bash
+# Install dependencies
+pip install -r requirements.txt
+
+# Run the service
+uvicorn app.main:app --host 0.0.0.0 --port 8005 --reload
+```
+
+## Development
+
+The service follows the standard microservice architecture:
+
+- `app/routers/` - API route handlers
+- `app/services/` - Business logic
+- `app/repositories/` - Data access layer
+- `app/models/` - Database models
+- `app/schemas/` - Pydantic schemas
+- `app/clients/` - Inter-service communication
+- `app/dependencies/` - FastAPI dependencies (auth, etc.)
+- `app/utils/` - Utility functions

app/app.py

@@ -1,33 +1,73 @@
+from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from insightfy_utils.logging import setup_logging, get_logger
 
 from app.routers.analytics_router import router as analytics_router
 #from app.routers.auth_route import router as auth_router
 
-app=FastAPI(
-    title="Analytics API",
+# Setup logging at module level
+setup_logging(level="INFO", format_type="json", app_name="insightfy-bloom-ms-ans")
+logger = get_logger(__name__)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan manager."""
+    # Startup
+    from app.sql import connect_to_database
+    from app.nosql import connect_stores
+    
+    logger.info("Starting ANS application")
+    await connect_to_database()
+    await connect_stores()
+    logger.info("Database connections established")
+    
+    yield
+    
+    # Shutdown
+    from app.sql import disconnect_from_database
+    from app.nosql import disconnect_stores
+    
+    logger.info("Shutting down ANS application")
+    await disconnect_from_database()
+    await disconnect_stores()
+    logger.info("Application shutdown complete")
+
+# Initialize FastAPI application
+app = FastAPI(
+    title="Analytics and Notification Service API",
     version="1.0",
-    description="API for analytics"
+    description="Analytics and Notification Service API for business intelligence and alerts",
+    lifespan=lifespan
 )
 
+# CORS configuration
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # Restrict to specific domains in production
+    allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-app.include_router(
-    analytics_router,
-    prefix="/api/v1/analytics",
-    tags=["Analytics"]
-)
+# Register routers
+app.include_router(analytics_router, prefix="/api/v1/analytics", tags=["Analytics"])
 
-# Optional root endpoint
+# Health check endpoints
 @app.get("/", tags=["Health"])
+async def root():
+    """Root endpoint - API information."""
+    return {
+        "service": "Analytics and Notification Service API",
+        "version": "1.0",
+        "status": "running"
+    }
+
+@app.get("/health", tags=["Health"])
 async def health_check():
-    """
-    Health check endpoint to verify the API is running.
-    """
-    return {"message": "Customer Analytics API is up and running!"}
+    """Health check endpoint."""
+    return {
+        "status": "healthy",
+        "service": "insightfy-bloom-ms-ans",
+        "version": "1.0"
+    }

app/cache.py

@@ -0,0 +1,23 @@
+from typing import Any
+from app.nosql import redis_client
+import json
+from insightfy_utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+CACHE_EXPIRY_SECONDS = 3600
+
+
+async def get_or_set_cache(key: str, fetch_func, expiry: int = CACHE_EXPIRY_SECONDS) -> Any:
+    """
+    Retrieve data from Redis cache or execute a function to fetch it.
+    """
+    cached_data = await redis_client.get(key)
+    if cached_data:
+        logger.info("Cache hit", extra={"key": key})
+        return json.loads(cached_data)
+
+    logger.info("Cache miss", extra={"key": key})
+    data = await fetch_func()
+    await redis_client.set(key, json.dumps(data), ex=expiry)
+    return data

app/clients/__init__.py

@@ -0,0 +1,7 @@
+"""Service clients for inter-service communication."""
+from app.clients.mpms_client import MPMSClient
+from app.clients.rms_client import RMSClient
+from app.clients.tms_client import TMSClient
+from app.clients.crm_client import CRMClient
+
+__all__ = ["MPMSClient", "RMSClient", "TMSClient", "CRMClient"]

app/clients/crm_client.py

@@ -0,0 +1,116 @@
+"""
+CRM (Customer Relationship Management) client for ANS.
+"""
+from insightfy_utils.http.clients import ServiceClient
+from insightfy_utils.logging import get_logger
+from typing import Optional, Dict, Any, List
+import os
+
+logger = get_logger(__name__)
+
+
+class CRMClient:
+    """Client for communicating with CRM."""
+    
+    def __init__(self):
+        base_url = os.getenv("CRM_BASE_URL", "http://localhost:8004")
+        self.client = ServiceClient(
+            base_url=base_url,
+            service_name="crm",
+            timeout=30.0,
+            max_retries=3
+        )
+    
+    async def get_customer_data(
+        self,
+        merchant_id: str,
+        customer_id: Optional[str] = None,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get customer data from CRM for analytics."""
+        try:
+            params = {"merchant_id": merchant_id}
+            if customer_id:
+                params["customer_id"] = customer_id
+                
+            response = await self.client.get(
+                "/api/v1/customers",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get customer data from CRM",
+                extra={
+                    "merchant_id": merchant_id,
+                    "customer_id": customer_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_campaign_data(
+        self,
+        merchant_id: str,
+        start_date: str,
+        end_date: str,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get campaign data from CRM for analytics."""
+        try:
+            params = {
+                "merchant_id": merchant_id,
+                "start_date": start_date,
+                "end_date": end_date
+            }
+            
+            response = await self.client.get(
+                "/api/v1/campaigns",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get campaign data from CRM",
+                extra={
+                    "merchant_id": merchant_id,
+                    "start_date": start_date,
+                    "end_date": end_date,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_loyalty_data(
+        self,
+        merchant_id: str,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get loyalty program data from CRM for analytics."""
+        try:
+            params = {"merchant_id": merchant_id}
+            
+            response = await self.client.get(
+                "/api/v1/loyalty",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get loyalty data from CRM",
+                extra={
+                    "merchant_id": merchant_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def close(self):
+        """Close the CRM client."""
+        await self.client.close()

app/clients/mpms_client.py

@@ -0,0 +1,104 @@
+"""
+MPMS (Merchant/Product Management Service) client for ANS.
+"""
+from insightfy_utils.http.clients import ServiceClient
+from insightfy_utils.logging import get_logger
+from typing import Optional, Dict, Any, List
+import os
+
+logger = get_logger(__name__)
+
+
+class MPMSClient:
+    """Client for communicating with MPMS."""
+    
+    def __init__(self):
+        base_url = os.getenv("MPMS_BASE_URL", "http://localhost:8001")
+        self.client = ServiceClient(
+            base_url=base_url,
+            service_name="mpms",
+            timeout=30.0,
+            max_retries=3
+        )
+    
+    async def get_product(
+        self,
+        merchant_id: str,
+        catalogue_id: str,
+        correlation_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Get product details from MPMS."""
+        try:
+            response = await self.client.get(
+                f"/api/v1/products/{catalogue_id}",
+                params={"merchant_id": merchant_id},
+                correlation_id=correlation_id
+            )
+            return response.json()
+        except Exception as e:
+            logger.error(
+                "Failed to get product from MPMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "catalogue_id": catalogue_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_products_batch(
+        self,
+        merchant_id: str,
+        catalogue_ids: List[str],
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get multiple products in batch from MPMS."""
+        try:
+            response = await self.client.post(
+                "/api/v1/products/batch",
+                json={
+                    "merchant_id": merchant_id,
+                    "catalogue_ids": catalogue_ids
+                },
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get products batch from MPMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "count": len(catalogue_ids),
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_merchant_info(
+        self,
+        merchant_id: str,
+        correlation_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Get merchant information from MPMS."""
+        try:
+            response = await self.client.get(
+                f"/api/v1/merchants/{merchant_id}",
+                correlation_id=correlation_id
+            )
+            return response.json()
+        except Exception as e:
+            logger.error(
+                "Failed to get merchant info from MPMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def close(self):
+        """Close the MPMS client."""
+        await self.client.close()

app/clients/rms_client.py

@@ -0,0 +1,109 @@
+"""
+RMS (Resource Management Service) client for ANS.
+"""
+from insightfy_utils.http.clients import ServiceClient
+from insightfy_utils.logging import get_logger
+from typing import Optional, Dict, Any
+import os
+
+logger = get_logger(__name__)
+
+
+class RMSClient:
+    """Client for communicating with RMS."""
+    
+    def __init__(self):
+        base_url = os.getenv("RMS_BASE_URL", "http://localhost:8002")
+        self.client = ServiceClient(
+            base_url=base_url,
+            service_name="rms",
+            timeout=30.0,
+            max_retries=3
+        )
+    
+    async def get_employee(
+        self,
+        merchant_id: str,
+        associate_id: str,
+        correlation_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Get employee details from RMS."""
+        try:
+            response = await self.client.get(
+                f"/api/v1/employees/{associate_id}",
+                params={"merchant_id": merchant_id},
+                correlation_id=correlation_id
+            )
+            return response.json()
+        except Exception as e:
+            logger.error(
+                "Failed to get employee from RMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "associate_id": associate_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_branch(
+        self,
+        merchant_id: str,
+        branch_id: str,
+        correlation_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Get branch details from RMS."""
+        try:
+            response = await self.client.get(
+                f"/api/v1/branches/{branch_id}",
+                params={"merchant_id": merchant_id},
+                correlation_id=correlation_id
+            )
+            return response.json()
+        except Exception as e:
+            logger.error(
+                "Failed to get branch from RMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "branch_id": branch_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def validate_permissions(
+        self,
+        merchant_id: str,
+        associate_id: str,
+        required_permissions: list,
+        correlation_id: Optional[str] = None
+    ) -> bool:
+        """Validate employee permissions in RMS."""
+        try:
+            response = await self.client.post(
+                "/api/v1/permissions/validate",
+                json={
+                    "merchant_id": merchant_id,
+                    "associate_id": associate_id,
+                    "permissions": required_permissions
+                },
+                correlation_id=correlation_id
+            )
+            return response.json().get("valid", False)
+        except Exception as e:
+            logger.error(
+                "Failed to validate permissions in RMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "associate_id": associate_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def close(self):
+        """Close the RMS client."""
+        await self.client.close()

app/clients/tms_client.py

@@ -0,0 +1,130 @@
+"""
+TMS (Transaction Management Service) client for ANS.
+"""
+from insightfy_utils.http.clients import ServiceClient
+from insightfy_utils.logging import get_logger
+from typing import Optional, Dict, Any, List
+import os
+
+logger = get_logger(__name__)
+
+
+class TMSClient:
+    """Client for communicating with TMS."""
+    
+    def __init__(self):
+        base_url = os.getenv("TMS_BASE_URL", "http://localhost:8003")
+        self.client = ServiceClient(
+            base_url=base_url,
+            service_name="tms",
+            timeout=30.0,
+            max_retries=3
+        )
+    
+    async def get_sales_data(
+        self,
+        merchant_id: str,
+        start_date: str,
+        end_date: str,
+        branch_id: Optional[str] = None,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get sales data from TMS for analytics."""
+        try:
+            params = {
+                "merchant_id": merchant_id,
+                "start_date": start_date,
+                "end_date": end_date
+            }
+            if branch_id:
+                params["branch_id"] = branch_id
+                
+            response = await self.client.get(
+                "/api/v1/sales",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get sales data from TMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "start_date": start_date,
+                    "end_date": end_date,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_stock_data(
+        self,
+        merchant_id: str,
+        branch_id: Optional[str] = None,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get stock data from TMS for analytics."""
+        try:
+            params = {"merchant_id": merchant_id}
+            if branch_id:
+                params["branch_id"] = branch_id
+                
+            response = await self.client.get(
+                "/api/v1/stock",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get stock data from TMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "branch_id": branch_id,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def get_appointment_data(
+        self,
+        merchant_id: str,
+        start_date: str,
+        end_date: str,
+        branch_id: Optional[str] = None,
+        correlation_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Get appointment data from TMS for analytics."""
+        try:
+            params = {
+                "merchant_id": merchant_id,
+                "start_date": start_date,
+                "end_date": end_date
+            }
+            if branch_id:
+                params["branch_id"] = branch_id
+                
+            response = await self.client.get(
+                "/api/v1/appointments",
+                params=params,
+                correlation_id=correlation_id
+            )
+            return response.json().get("data", [])
+        except Exception as e:
+            logger.error(
+                "Failed to get appointment data from TMS",
+                extra={
+                    "merchant_id": merchant_id,
+                    "start_date": start_date,
+                    "end_date": end_date,
+                    "correlation_id": correlation_id
+                },
+                exc_info=e
+            )
+            raise
+    
+    async def close(self):
+        """Close the TMS client."""
+        await self.client.close()

app/dependencies/auth.py

@@ -1,22 +1,26 @@
-import logging
+from insightfy_utils.logging import get_logger
 from fastapi import Depends, HTTPException, status
 from enum import Enum
-from app.util.jwt import decode_jwt_token
+from app.utils.jwt import decode_jwt_token
 from settings import SECRET_KEY, ALGORITHM
 from fastapi.security import APIKeyHeader
 from app.nosql import mongo_db
 
-# Logger setup (consistent with project)
-logger = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
-# AccessID Enum for permission strings
-class AccessID(str, Enum):
-    # Catalogue permissions
-    CREATE_CUSTOMER = "create_customers"
-    UPDATE_CUSTOMER = "update_customers"
-    VIEW_CUSTOMER = "view_customers"
-    DELETE_CUSTOMER = "delete_customers"
-    
+class AccessID(Enum):
+    VIEW_ANALYTICS = "view_analytics"
+    CREATE_ANALYTICS = "create_analytics"
+    UPDATE_ANALYTICS = "update_analytics"
+    DELETE_ANALYTICS = "delete_analytics"
+    VIEW_REPORTS = "view_reports"
+    CREATE_REPORTS = "create_reports"
+    UPDATE_REPORTS = "update_reports"
+    DELETE_REPORTS = "delete_reports"
+    VIEW_DASHBOARD = "view_dashboard"
+    CREATE_DASHBOARD = "create_dashboard"
+    UPDATE_DASHBOARD = "update_dashboard"
+    DELETE_DASHBOARD = "delete_dashboard"
 
 oauth2_scheme = APIKeyHeader(name="Authorization")
 
@@ -25,35 +29,34 @@ def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
         if token.startswith("Bearer "):
             token = token.split(" ")[1]
         payload = decode_jwt_token(token)
-        logger.info(f"Authenticated user {payload.get('associate_id')} for merchant {payload.get('merchant_id')}")
+        logger.info("User authenticated", extra={
+            "associate_id": payload.get("associate_id"),
+            "merchant_id": payload.get("merchant_id")
+        })
         return {
             "associate_id": payload["associate_id"],
             "merchant_id": payload["merchant_id"],
-            "branch_id":payload["branch_id"],
-            "role": payload.get("role", "user")
+            "branch_id": payload["branch_id"],
+            "role_id": payload.get("role_id", "user")
         }
     except Exception as e:
-        logger.warning(f"Authentication failed: {str(e)}")
+        logger.warning("Authentication failed", exc_info=e)
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid authentication credentials",
         )
-    
 
 async def user_has_permission(merchant_id: str, role_id: str, permission: str) -> bool:
-    """
-    Check if the user has the required permission using MongoDB filtering.
-    permission format: RESOURCE:ACTION (e.g., appointments:create)
-    """
+    """Check if the user has the required permission using MongoDB filtering."""
     try:
-        # Support both 'RESOURCE:ACTION' and 'ACTION_RESOURCE' formats
         if "_" in permission:
             parts = permission.split("_", 1)
             if len(parts) == 2:
                 action, resource = parts
             else:
-                logger.warning(f"Permission format invalid: {permission}")
+                logger.warning("Invalid permission format", extra={"permission": permission})
                 return False
+        
         query = {
             "merchant_id": merchant_id,
             "role_id": role_id,
@@ -61,30 +64,40 @@ async def user_has_permission(merchant_id: str, role_id: str, permission: str) -
         }
         role_doc = await mongo_db["access_roles"].find_one(query)
         has_perm = role_doc is not None
-        logger.info(f"Permission check for role {role_id}, merchant {merchant_id}, permission {permission}: {has_perm}")
+        logger.info("Permission check", extra={
+            "role_id": role_id,
+            "merchant_id": merchant_id,
+            "permission": permission,
+            "granted": has_perm
+        })
         return has_perm
     except Exception as e:
-        logger.error(f"Permission check error: {str(e)}")
+        logger.error("Permission check error", exc_info=e)
         return False
 
-# Reusable FastAPI dependency for permission checks
 async def require_permission(
     access_id: str,
     current_user: dict = Depends(get_current_user)
 ) -> dict:
-    """
-    Dependency to check user permission for a given access_id (permission string).
-    Raises HTTPException(403) if not permitted.
-    Returns current_user if permitted.
-    """
+    """Dependency to check user permission for a given access_id."""
     merchant_id = current_user.get("merchant_id")
     user_id = current_user.get("associate_id")
-    role_id = current_user.get("role")
+    role_id = current_user.get("role_id")
+
+    logger.info("Checking permission", extra={
+        "user_id": user_id,
+        "merchant_id": merchant_id,
+        "role_id": role_id,
+        "access_id": access_id
+    })
+
     if not user_id or not merchant_id or not role_id:
-        logger.warning(f"Permission denied: missing user/merchant/role info for user {user_id}")
+        logger.warning("Permission denied: missing credentials", extra={"user_id": user_id})
         raise HTTPException(status_code=403, detail="Forbidden")
+    
     if not await user_has_permission(merchant_id, role_id, access_id):
-        logger.warning(f"Permission denied for user {user_id} on {access_id}")
+        logger.warning("Permission denied", extra={"user_id": user_id, "access_id": access_id})
         raise HTTPException(status_code=403, detail="Forbidden")
-    logger.info(f"Permission granted for user {user_id} on {access_id}")
-    return current_user
+    
+    logger.info("Permission granted", extra={"user_id": user_id, "access_id": access_id})
+    return current_user

app/main.py

@@ -0,0 +1,142 @@
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from contextlib import asynccontextmanager
+
+from insightfy_utils.logging import get_logger
+from insightfy_utils.telemetry.health import HealthCheck, HealthStatus
+from insightfy_utils.telemetry.metrics import MetricsCollector, track_request_metrics
+from insightfy_utils.models.shared_schemas import HealthCheckResponse, HealthCheckComponent
+
+from app.sql import database
+from app.nosql import redis_client
+from app.clients import MPMSClient, RMSClient, TMSClient, CRMClient
+
+logger = get_logger(__name__)
+
+# Initialize health check and metrics
+health_check = HealthCheck(service_name="ans")
+metrics_collector = MetricsCollector(service_name="ans")
+
+# Initialize service clients
+mpms_client = MPMSClient()
+rms_client = RMSClient()
+tms_client = TMSClient()
+crm_client = CRMClient()
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan events"""
+    # Startup
+    logger.info("Starting ANS service")
+    
+    try:
+        # Test database connections
+        await database.connect()
+        health_check.add_component("postgresql", HealthStatus.HEALTHY)
+        logger.info("PostgreSQL connection established")
+    except Exception as e:
+        logger.error("Failed to connect to PostgreSQL", exc_info=e)
+        health_check.add_component("postgresql", HealthStatus.UNHEALTHY, str(e))
+    
+    try:
+        # Test Redis connection
+        await redis_client.ping()
+        health_check.add_component("redis", HealthStatus.HEALTHY)
+        logger.info("Redis connection established")
+    except Exception as e:
+        logger.error("Failed to connect to Redis", exc_info=e)
+        health_check.add_component("redis", HealthStatus.UNHEALTHY, str(e))
+    
+    yield
+    
+    # Shutdown
+    logger.info("Shutting down ANS service")
+    await database.disconnect()
+    await mpms_client.close()
+    await rms_client.close()
+    await tms_client.close()
+    await crm_client.close()
+    logger.info("Service clients closed")
+
+app = FastAPI(
+    title="Analytics and Notification Service",
+    description="ANS - Handles analytics, reporting, and notifications",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Add metrics middleware
+@app.middleware("http")
+async def metrics_middleware(request: Request, call_next):
+    """Track request metrics"""
+    return await track_request_metrics(
+        request=request,
+        call_next=call_next,
+        metrics_collector=metrics_collector,
+        logger=logger
+    )
+
+# Make clients available to the app
+app.state.mpms_client = mpms_client
+app.state.rms_client = rms_client
+app.state.tms_client = tms_client
+app.state.crm_client = crm_client
+
+# Health check endpoint
+@app.get("/health", response_model=HealthCheckResponse, tags=["Health"])
+async def health():
+    """Service health check endpoint"""
+    health_data = health_check.get_health()
+    
+    # Convert to standardized response format
+    components = {}
+    for name, component in health_data.get("components", {}).items():
+        components[name] = HealthCheckComponent(
+            status=component.get("status"),
+            message=component.get("message")
+        )
+    
+    return HealthCheckResponse(
+        status=health_data["status"],
+        service="ans",
+        version="1.0.0",
+        components=components
+    )
+
+# Readiness check endpoint
+@app.get("/ready", tags=["Health"])
+async def ready():
+    """Service readiness check endpoint"""
+    health_data = health_check.get_health()
+    
+    # Service is ready if all critical components are healthy
+    is_ready = health_data["status"] == "healthy"
+    
+    if is_ready:
+        return JSONResponse(content=health_data, status_code=200)
+    else:
+        return JSONResponse(content=health_data, status_code=503)
+
+# Liveness check endpoint
+@app.get("/live", tags=["Health"])
+async def live():
+    """Service liveness check endpoint"""
+    return {"status": "alive", "service": "ans"}
+
+# Metrics endpoint
+@app.get("/metrics", tags=["Metrics"])
+async def metrics():
+    """Prometheus-compatible metrics endpoint"""
+    return metrics_collector.get_metrics()
+
+# ...existing code... (include routers)

app/model/appointment_analytics.py

@@ -1,71 +0,0 @@
-from sqlalchemy import (
-    Column,
-    Integer,
-    String,
-    Date,
-    Time,
-    ForeignKey,
-    DECIMAL,
-    TIMESTAMP,
-    text,
-    #Index,
-)
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import relationship
-
-# Create a base class for declarative models
-Base = declarative_base()
-
-class AppointmentAnalytics(Base):
-    """
-    SQLAlchemy ORM model for the `appointment_analytics` table.
-    """
-    __tablename__ = "appointment_analytics"
-
-    # Columns
-    appointment_id = Column(String(50), primary_key=True)
-    customer_id = Column(String(50), ForeignKey("customer_analytics.customer_id"), nullable=False)
-    merchant_id = Column(String(50), nullable=False)
-    location_id = Column(String(20), nullable=False)
-    staff_id = Column(String(50))
-    appointment_date = Column(Date, nullable=False)
-    appointment_time = Column(Time)
-    booking_date = Column(Date)
-    days_advance_booking = Column(Integer)
-    service_name = Column(String(200))
-    service_category = Column(String(100))
-    duration_minutes = Column(Integer)
-    service_price = Column(DECIMAL(10, 2))
-    appointment_status = Column(String(20))
-    cancellation_reason = Column(String(200))
-    no_show_reason = Column(String(200))
-    revenue = Column(DECIMAL(10, 2), default=0)
-    discount_applied = Column(DECIMAL(10, 2), default=0)
-    payment_method = Column(String(50))
-    wait_time_minutes = Column(Integer)
-    satisfaction_rating = Column(DECIMAL(3, 2))
-    created_at = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"))
-    updated_at = Column(
-        TIMESTAMP,
-        server_default=text("CURRENT_TIMESTAMP"),
-        onupdate=text("CURRENT_TIMESTAMP"),
-    )
-
-    # Relationships (with `customer_analytics` model )
-    customer = relationship("CustomerAnalytics", back_populates="appointments")
-    ''' 
-    # Table arguments for indexes
-    __table_args__ = (
-        Index("idx_appointment_date", appointment_date),
-        Index("idx_appointment_customer", customer_id),
-        Index("idx_appointment_status", appointment_status),
-        Index("idx_appointment_service", service_category),
-    )
-    '''
-     
-    def __repr__(self):
-        return (
-            f"<AppointmentAnalytics(appointment_id='{self.appointment_id}', "
-            f"customer_id='{self.customer_id}', "
-            f"appointment_date='{self.appointment_date}')>"
-        )

app/model/customer_analytics.py

@@ -1,56 +0,0 @@
-from sqlalchemy import Column, Integer, String, Date, TIMESTAMP, Numeric, Text
-from sqlalchemy.orm import declarative_base, relationship
-
-# Create a base class for declarative models
-Base = declarative_base()
-
-class CustomerAnalytics(Base):
-    __tablename__ = 'customer_analytics'
-
-    customer_id = Column(String(50), primary_key=True, nullable=False)
-    merchant_id = Column(String(50), nullable=False)
-    location_id = Column(String(20), nullable=False)
-    customer_name = Column(String(200))
-    email = Column(String(200))
-    phone = Column(String(20))
-    city = Column(String(100))
-    state = Column(String(100))
-    country = Column(String(100))
-    acquisition_date = Column(Date, nullable=False)
-    acquisition_source = Column(String(100))
-    acquisition_channel = Column(String(50))
-    campaign_id = Column(String(50))
-    referral_customer_id = Column(String(50))
-    acquisition_cost = Column(Numeric(10, 2), default=0)
-    first_appointment_date = Column(Date)
-    days_to_first_appointment = Column(Integer)
-    total_revenue = Column(Numeric(10, 2), default=0)
-    total_visits = Column(Integer, default=0)
-    avg_spend_per_visit = Column(Numeric(10, 2), default=0)
-    customer_lifetime_value = Column(Numeric(12, 2), default=0)
-    last_visit_date = Column(Date)
-    last_interaction_date = Column(Date)
-    loyalty_points = Column(Integer, default=0)
-    membership_type = Column(String(50))
-    subscription_plan = Column(String(100))
-    visit_frequency_days = Column(Numeric(5, 1))
-    churn_risk_score = Column(Numeric(3, 2), default=0)
-    churn_risk_level = Column(String(20), default='Low')
-    customer_status = Column(String(20), default='Active')
-    avg_satisfaction_score = Column(Numeric(3, 2))
-    nps_score = Column(Integer)
-    total_surveys_completed = Column(Integer, default=0)
-    last_survey_date = Column(Date)
-    created_at = Column(TIMESTAMP(timezone=True), server_default='now()')
-    updated_at = Column(TIMESTAMP(timezone=True), server_default='now()', onupdate='now()')
-
-    # This creates the link from a customer to all their appointments
-    # 'appointments' is the name of this attribute on this model
-    appointments = relationship(
-        "AppointmentAnalytics",
-        back_populates="customer",
-        cascade="all, delete-orphan" #for managing related objects
-    )
-    
-    def __repr__(self):
-        return f"<Customer(id={self.customer_id}, name='{self.customer_name}', email='{self.email}')>"

app/models/__init__.py

@@ -0,0 +1 @@
+"""Database models for ANS service."""

app/nosql.py

@@ -1,14 +1,10 @@
 # data_stores.py (or your current filename)
 from __future__ import annotations
 
-import logging
-from urllib.parse import urlparse
+from insightfy_utils.logging import get_logger
+from insightfy_utils.db.mongo_connector import create_mongo_connection
+from insightfy_utils.db.redis_connector import create_redis_connection
 
-import motor.motor_asyncio
-import redis.asyncio as redis
-from redis.exceptions import RedisError
-
-# Single source of truth
 from settings import (
     MONGO_URI,
     MONGO_DB_NAME,
@@ -18,14 +14,7 @@ from settings import (
     CACHE_DB,
 )
 
-# -----------------------------------------------------------------------------
-# Logging
-# -----------------------------------------------------------------------------
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-)
-logger = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 # -----------------------------------------------------------------------------
 # Validate configuration
@@ -41,45 +30,39 @@ if not CACHE_URL and not CACHE_URI:
     )
 
 # -----------------------------------------------------------------------------
-# MongoDB client (motor)
+# MongoDB client
 # -----------------------------------------------------------------------------
 try:
-    mongo_client = motor.motor_asyncio.AsyncIOMotorClient(
+    mongo_client, mongo_db = create_mongo_connection(
         MONGO_URI,
-        serverSelectionTimeoutMS=60_000,
-        socketTimeoutMS=60_000,
-        connectTimeoutMS=60_000,
+        MONGO_DB_NAME,
+        server_selection_timeout_ms=60_000,
+        socket_timeout_ms=60_000,
+        connect_timeout_ms=60_000,
     )
-    mongo_db = mongo_client[MONGO_DB_NAME]
-    logger.info("MongoDB client initialized (db=%s).", MONGO_DB_NAME)
+    logger.info("MongoDB client initialized", extra={"db_name": MONGO_DB_NAME})
 except Exception as e:
-    logger.exception("Failed to initialize MongoDB client.")
+    logger.error("Failed to initialize MongoDB client", exc_info=e)
     raise
 
 # -----------------------------------------------------------------------------
-# Redis client (async)
+# Redis client
 # -----------------------------------------------------------------------------
 
 
-def _redis_from_settings() -> redis.Redis:
-    """
-    Prefer a single URL (CACHE_URL / REDIS_URL). If not provided,
-    fall back to CACHE_URI ("host:port") + optional password.
-    """
+def _redis_from_settings():
+    """Create Redis client from settings."""
     if CACHE_URL:
-        # Supports redis:// and rediss:// with inline password and db index
+        import redis.asyncio as redis
         return redis.from_url(CACHE_URL, decode_responses=True)
 
-    # Fallback: host:port (required), password optional
     try:
         host, port_str = CACHE_URI.split(":")
         port = int(port_str)
     except Exception:
-        raise ValueError(
-            "Invalid CACHE_URI format. Expected 'host:port', e.g., 'localhost:6379'."
-        )
+        raise ValueError("Invalid CACHE_URI format. Expected 'host:port'")
 
-    return redis.Redis(
+    return create_redis_connection(
         host=host,
         port=port,
         password=CACHE_PASSWORD or None,
@@ -90,31 +73,33 @@ def _redis_from_settings() -> redis.Redis:
 
 try:
     redis_client = _redis_from_settings()
-    logger.info("Redis client initialized.")
+    logger.info("Redis client initialized")
 except Exception as e:
-    logger.exception("Failed to initialize Redis client.")
+    logger.error("Failed to initialize Redis client", exc_info=e)
     raise
 
 # -----------------------------------------------------------------------------
-# Optional: quick health helpers
+# Health check helpers
 # -----------------------------------------------------------------------------
 
 
 async def ping_mongo() -> bool:
+    """Check MongoDB connectivity."""
     try:
         await mongo_db.command("ping")
         return True
     except Exception:
-        logger.exception("Mongo ping failed.")
+        logger.error("Mongo ping failed", exc_info=True)
         return False
 
 
 async def ping_redis() -> bool:
+    """Check Redis connectivity."""
     try:
         pong = await redis_client.ping()
         return bool(pong)
-    except RedisError:
-        logger.exception("Redis ping failed.")
+    except Exception:
+        logger.error("Redis ping failed", exc_info=True)
         return False
 
 # -----------------------------------------------------------------------------
@@ -123,30 +108,25 @@ async def ping_redis() -> bool:
 
 
 async def connect_stores() -> None:
-    """
-    No-op for motor/redis (lazy connections), but you can force a ping to fail fast.
-    """
+    """Verify store connectivity on startup."""
     ok_mongo = await ping_mongo()
     ok_redis = await ping_redis()
     if not (ok_mongo and ok_redis):
         raise RuntimeError("Store connectivity check failed (mongo or redis).")
-    logger.info("Store connectivity OK (Mongo & Redis).")
+    logger.info("Store connectivity OK (Mongo & Redis)")
 
 
 async def disconnect_stores() -> None:
-    """
-    Close drivers on shutdown.
-    """
+    """Close connections on shutdown."""
     try:
         mongo_client.close()
     except Exception:
-        logger.exception("Error closing Mongo client.")
+        logger.error("Error closing Mongo client", exc_info=True)
 
     try:
         await redis_client.close()
     except Exception:
-        logger.exception("Error closing Redis client.")
+        logger.error("Error closing Redis client", exc_info=True)
 
 # Alias for backward compatibility
-db = mongo_db
-
+db = mongo_db

app/repositories/__init__.py

@@ -0,0 +1 @@
+"""Repository layer for data access in ANS service."""

app/routers/__init__.py

@@ -0,0 +1 @@
+"""API routers for ANS service."""

app/routers/analytics_router.py

@@ -1,125 +1,77 @@
-import logging
-import time
-from typing import Optional 
-from app.services.customer_analytics_service import CustomerAnalyticService
-from app.services.appointment_analytics_service import AppointmentAnalyticService 
-from app.dependencies.auth import AccessID, get_current_user, require_permission
-from fastapi import APIRouter, Body, Depends, FastAPI, HTTPException, Path, Query
-
-router=APIRouter()
-logger=logging.getLogger(__name__)
-
-INTERNAL_SERVER_ERROR = "Internal server error"
-ACCESS_DENIED_ERROR = "Access denied"
-ADMIN_MANAGER_ONLY_ERROR = "Only Admin or branch manager can delete items"
-
-
-async def require_view_customer_permission(current_user: dict = Depends(get_current_user)):
-    return await require_permission(AccessID.VIEW_CUSTOMER.value, current_user)
-
- 
-'''
----------- CUSTOMER ANALYTICS APIS -----------------------
-'''
-@router.get("/list", status_code=200)
-async def list_customer_analytics(current_user: dict = Depends(require_view_customer_permission)):
-    
-    print("list_customer_analytics") 
-    merchant_id = current_user.get("merchant_id")
-    result = await CustomerAnalyticService.get_all_customer_analytics(merchant_id)
-    print("result ",result)
-    if(result is not None):
-        return {**result}
-    else:
-        return {"error": "Customers not found"}
-
-@router.get("/{customer_id}", status_code=200)
-async def get_by_customer_id(customer_id: str,
-                             current_user: dict = Depends(require_view_customer_permission)):
-    print("get_by_customer_id") 
-    
-    merchant_id = current_user.get("merchant_id")
-    result = await CustomerAnalyticService.get_by_customer_id(customer_id, merchant_id)
-    print("result ",result)
-    
-    if(result is not None):
-        return {**result}
-    else:
-        return {"error": "Customers not found"}
-
-@router.get("/merchant_id/{merchant_id}", status_code=200)
-async def get_customers_by_merchant_id(merchant_id: str,
-                             current_user: dict = Depends(require_view_customer_permission)):
-    print("get_by_merchant_id") 
-    
-    curuser_merchant_id = current_user.get("merchant_id")
-    
-    if(curuser_merchant_id==merchant_id):
-        result = await CustomerAnalyticService.get_by_merchant_id(merchant_id)
-        print("result ",result)
-    else:
-        return {"error": "Customers not found"}
-        #raise HTTPException(status_code=403, detail=ACCESS_DENIED_ERROR)
-
-    return {**result}
-
-
-'''
-------------------APPOINTMENT ANALYTCIS APIS--------------------
-'''
-@router.get("/appointment/list/{merchant_id}", status_code=200)
-async def list_appointment_analytics(merchant_id: str,
-                             current_user: dict = Depends(require_view_customer_permission)):
-    print("list_appointment_analytics") 
-    
-    curuser_merchant_id = current_user.get("merchant_id")
-    
-    if(curuser_merchant_id==merchant_id):
-        result = await AppointmentAnalyticsService.get_all_appointment_analytics(merchant_id)
-        print("result ",result)
-    else:
-        return {"error": "Appointments not found"}
-        #raise HTTPException(status_code=403, detail=ACCESS_DENIED_ERROR)
-
-    return {**result}
-
-@router.get("/appointment/customer/{customer_id}", status_code=200)
-async def get_appointments_by_customerid(customer_id: str,
-                             current_user: dict = Depends(require_view_customer_permission)):
-    print("get_appointments_by_customerid") 
-    
-    merchant_id = current_user.get("merchant_id")
-    
-    result = await AppointmentAnalyticsService.get_appointments_by_customerid(customer_id,merchant_id)
+"""
+Analytics router for ANS service.
+"""
+from fastapi import APIRouter, Depends, HTTPException
+from typing import List, Optional
+from datetime import datetime, date
+
+from app.dependencies.auth import get_current_user, require_permission, AccessID
+from insightfy_utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+router = APIRouter()
+
+@router.get("/health")
+async def analytics_health():
+    """Analytics service health check."""
+    return {"status": "healthy", "service": "analytics"}
+
+@router.get("/dashboard")
+async def get_dashboard_data(
+    start_date: Optional[date] = None,
+    end_date: Optional[date] = None,
+    current_user: dict = Depends(get_current_user)
+):
+    """Get dashboard analytics data."""
+    merchant_id = current_user["merchant_id"]
     
-    if(result is not None):
-        #return {**result}
-        print("result ",result)
-    else:
-        return {"error": "Appointments not found"}
-        #raise HTTPException(status_code=403, detail=ACCESS_DENIED_ERROR)
-
-    return {**result}
-
-
- 
-
-@router.get("/appointment/{appointment_id}", status_code=200)
-async def get_appointments_by_id(appointment_id: str,
-                             current_user: dict = Depends(require_view_customer_permission)):
-    print("get_appointments_by_id") 
+    logger.info("Getting dashboard data", extra={
+        "merchant_id": merchant_id,
+        "start_date": start_date,
+        "end_date": end_date
+    })
     
-    merchant_id = current_user.get("merchant_id")
+    # Placeholder implementation
+    return {
+        "merchant_id": merchant_id,
+        "data": {
+            "sales_summary": {},
+            "customer_metrics": {},
+            "inventory_status": {},
+            "performance_indicators": {}
+        },
+        "period": {
+            "start_date": start_date,
+            "end_date": end_date
+        }
+    }
+
+@router.get("/reports")
+async def get_reports(
+    report_type: str,
+    start_date: Optional[date] = None,
+    end_date: Optional[date] = None,
+    current_user: dict = Depends(require_permission(AccessID.VIEW_REPORTS.value))
+):
+    """Get analytics reports."""
+    merchant_id = current_user["merchant_id"]
     
-    result = await AppointmentAnalyticsService.get_by_appointment_id(appointment_id,merchant_id)
+    logger.info("Getting reports", extra={
+        "merchant_id": merchant_id,
+        "report_type": report_type,
+        "start_date": start_date,
+        "end_date": end_date
+    })
     
-    if(result is not None):
-        #return {**result}
-        print("result ",result)
-    else:
-        return {"error": "Appointments not found"}
-        #raise HTTPException(status_code=403, detail=ACCESS_DENIED_ERROR)
-
-    return {**result}
-
-
+    # Placeholder implementation
+    return {
+        "merchant_id": merchant_id,
+        "report_type": report_type,
+        "data": [],
+        "generated_at": datetime.utcnow(),
+        "period": {
+            "start_date": start_date,
+            "end_date": end_date
+        }
+    }

app/routers/auth_route.py

@@ -1,30 +0,0 @@
-from fastapi import APIRouter, HTTPException, status, Depends, Form
-from jose import jwt
-from datetime import datetime, timedelta
-from settings import SECRET_KEY, ALGORITHM
-
-router = APIRouter(prefix="/auth", tags=["Authentication"])
-
-# Hardcoded test user
-USERS_DB = {
-    "admin@example.com": {
-        "password": "admin123",
-        "merchant_id": "MERCHANT001",
-        "associate_id": "admin"
-    }
-}
-
-@router.post("/login")
-def login(username: str = Form(...), password: str = Form(...)):
-    user = USERS_DB.get(username)
-    if not user or user["password"] != password:
-        raise HTTPException(status_code=401, detail="Invalid credentials")
-
-    token_data = {
-        "sub": user["associate_id"],
-        "merchant_id": user["merchant_id"],
-        "exp": datetime.utcnow() + timedelta(hours=2)
-    }
-
-    access_token = jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)
-    return {"access_token": access_token, "token_type": "bearer"}

app/schemas/__init__.py

@@ -0,0 +1 @@
+"""Pydantic schemas for ANS service."""

app/services/__init__.py

@@ -0,0 +1 @@
+"""Business logic services for ANS service."""

app/services/appointment_analytics_service.py

@@ -1,76 +0,0 @@
-from app.model.appointment_analytics import AppointmentAnalytics
-from sqlalchemy import select, and_
-from app.util.dbutil import get_db_session
-import logging 
-
-logger=logging.getLogger(__name__)
-
-class AppointmentAnalyticService:
-    @staticmethod
-    async def get_all_appointment_analytics(merchantid):
-        
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                result = await session.execute(select(AppointmentAnalytics).filter_by(merchant_id=merchantid))
-                appointments = result.scalars().all()
-                
-                result_dict = {"total": len(appointments), 
-                               "appointments":appointments}
-                return result_dict 
-
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-    
-    @staticmethod
-    async def get_appointments_by_customerid(customerid,merchantid):
-        
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                #customer = session.query(AppointmentAnalytics).filter_by(merchant_id=merchantid).first()
-                result =  await session.execute(select(AppointmentAnalytics)
-                .filter(
-                    and_(
-                        AppointmentAnalytics.customer_id == customerid,
-                        AppointmentAnalytics.merchant_id == merchantid
-                    )
-                )
-                )
-                appointments = result.scalars().all()
-                
-                result_dict = {"total": len(appointments), 
-                               "appointments":appointments}
-                return result_dict 
-
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-      
-    @staticmethod
-    async def get_by_appointment_id(appointmentid, merchantid):
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                #customer = session.query(AppointmentAnalytics).filter_by(merchant_id=merchantid).first()
-                result =  await session.execute(select(AppointmentAnalytics)
-                .filter(
-                    and_(
-                        AppointmentAnalytics.appointment_id == appointmentid,
-                        AppointmentAnalytics.merchant_id == merchantid
-                    )
-                )
-                )
-                
-                appointment = result.scalars().first()
-                
-                result_dict = {"appointment" : "Appointments not found"}
-                if appointment:
-                    result_dict = {"appointment": appointment}
-                return result_dict
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-        
-        

app/services/customer_analytics_service.py

@@ -1,71 +0,0 @@
-from app.model.customer_analytics import CustomerAnalytics
-from sqlalchemy import select, and_
-from app.util.dbutil import get_db_session
-import logging 
-
-logger=logging.getLogger(__name__)
-
-class CustomerAnalyticService:
-    
-    @staticmethod
-    async def get_all_customer_analytics(merchantid):
-        
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                #customer = session.query(CustomerAnalytics).all() 
-                #result = await session.execute(select(CustomerAnalytics))
-                result = await session.execute(select(CustomerAnalytics).filter_by(merchant_id=merchantid))
-                customers = result.scalars().all()
-                
-                result_dict = {"total": len(customers), 
-                               "customers":customers}
-                return result_dict 
-
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-        
-    @staticmethod
-    async def get_by_customer_id(customerid, merchantid):
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                #customer = session.query(CustomerAnalytics).filter_by(customer_id=customerid).first()
-                #result =  await session.execute(select(CustomerAnalytics).filter_by(customer_id=customerid))
-                result =  await session.execute(select(CustomerAnalytics)
-                .filter(
-                    and_(
-                        CustomerAnalytics.customer_id == customerid,
-                        CustomerAnalytics.merchant_id == merchantid
-                    )
-                )
-                )
-                
-                customer = result.scalars().first()
-                
-                result_dict = {"customer" : "Customer not found"}
-                if customer:
-                    result_dict = {"customer": customer}
-                return result_dict
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-        
-    @staticmethod
-    async def get_by_merchant_id(merchantid):
-        try:
-            # The 'with' statement handles session management
-            async with get_db_session() as session:
-                #customer = session.query(CustomerAnalytics).filter_by(customer_id=customerid).first()
-                result =  await session.execute(select(CustomerAnalytics).filter_by(merchant_id=merchantid))
-                customers = result.scalars().all()
-                
-                result_dict = {"customers" : "Customer not found"}
-                if customers:
-                    result_dict = {"customers": customers}
-                return result_dict
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None
-        

app/sql.py

@@ -0,0 +1,67 @@
+# database.py
+from __future__ import annotations
+
+from insightfy_utils.logging import get_logger
+from insightfy_utils.db.postgres_connector import create_postgres_connection, create_postgres_metadata
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+from settings import DATABASE_URI
+
+logger = get_logger(__name__)
+
+logger.info("Using DATABASE_URL", extra={"url": DATABASE_URI})
+
+if not DATABASE_URI:
+    logger.error("DATABASE_URI is empty or missing from settings")
+    raise ValueError("DATABASE_URI is not set. Check environment variables.")
+
+# Use centralized database connection from insightfy-utils
+database = create_postgres_connection(DATABASE_URI)
+metadata = create_postgres_metadata()
+
+logger.info("Database configuration loaded successfully")
+
+# ────────────────────────────────────────────────────────────────────────────────
+# Lifecycle helpers
+# ────────────────────────────────────────────────────────────────────────────────
+async def connect_to_database() -> None:
+    """Connects to the database when the application starts."""
+    try:
+        await database.connect()
+        logger.info("Successfully connected to the database")
+    except Exception as e:
+        logger.exception("Error connecting to the database")
+        raise
+
+async def disconnect_from_database() -> None:
+    """Disconnects from the database when the application shuts down."""
+    try:
+        await database.disconnect()
+        logger.info("Successfully disconnected from the database")
+    except Exception as e:
+        logger.exception("Error disconnecting from the database")
+        raise
+
+# For advanced use cases with custom connection pool settings
+async_engine = create_async_engine(
+    DATABASE_URI, 
+    echo=False, 
+    future=True,
+    pool_size=10,
+    max_overflow=20,
+    pool_timeout=30,
+    pool_recycle=3600,
+    pool_pre_ping=True,
+    connect_args={
+        "server_settings": {
+            "application_name": "insightfy-ans",
+            "jit": "off"
+        },
+        "command_timeout": 60,
+        "statement_cache_size": 0
+    }
+)
+async_session = sessionmaker(
+    async_engine, expire_on_commit=False, class_=AsyncSession
+)

app/util/dbutil.py

@@ -1,48 +0,0 @@
-#from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
-from contextlib import contextmanager, asynccontextmanager
-#import asyncio
-
-from settings import DB_PROTOCOL, DB_USER,DB_PASSWORD,DB_HOST,DB_PORT,DB_NAME,DB_SSLMODE
-# 1. Use a single, global Session factory
-# This factory should only be created once when your application starts.
-
-
-#db_url = '{protocol}://{user}:{password}@{host}:{port}/{name}?sslmode={sslmode}'.format(
-async_db_url = '{protocol}://{user}:{password}@{host}:{port}/{name}?ssl={sslmode}'.format(
-    protocol=DB_PROTOCOL,
-    user=DB_USER,
-    password=DB_PASSWORD,
-    host=DB_HOST,
-    port=DB_PORT,
-    name=DB_NAME,
-    sslmode = "require" if DB_SSLMODE=="require" else None
-    
-    #sslmode=DB_SSLMODE
-)
-
-#print("async_db_url = ",async_db_url)
-#engine = create_engine(db_url)
-async_engine = create_async_engine(async_db_url)
-#session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-asyncsession = sessionmaker(
-    async_engine, expire_on_commit=False, class_=AsyncSession
-)
-
-# 2. Use a context manager to manage sessions
-#@contextmanager
-@asynccontextmanager 
-async def get_db_session():
-    """
-    Provides a transactional scope around a series of operations.
-    A new session is created, used, and then automatically closed.
-    """
-    db = asyncsession()
-    try:
-        yield db
-    finally:
-        #db.close()
-        await db.close()
-
-    

app/utils/db_utils.py

@@ -0,0 +1,126 @@
+"""
+Database utility functions for connection management and retry logic.
+"""
+import asyncio
+from typing import Any, Callable, TypeVar
+from functools import wraps
+from sqlalchemy.exc import DisconnectionError, InterfaceError, OperationalError
+from app.sql import async_session, async_engine
+from insightfy_utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+# Type variable for return type
+T = TypeVar('T')
+
+class DatabaseConnectionError(Exception):
+    """Custom exception for database connection issues."""
+    pass
+
+def retry_db_operation(max_retries: int = 3, delay: float = 1.0):
+    """
+    Decorator to retry database operations on connection failures.
+    
+    Args:
+        max_retries: Maximum number of retry attempts
+        delay: Delay between retries in seconds
+    """
+    def decorator(func: Callable[..., T]) -> Callable[..., T]:
+        @wraps(func)
+        async def wrapper(*args, **kwargs) -> T:
+            last_exception = None
+            
+            for attempt in range(max_retries + 1):
+                try:
+                    return await func(*args, **kwargs)
+                except (DisconnectionError, InterfaceError, OperationalError) as e:
+                    last_exception = e
+                    logger.warning(
+                        f"Database operation failed (attempt {attempt + 1}/{max_retries + 1}): {e}"
+                    )
+                    
+                    if attempt < max_retries:
+                        # Exponential backoff
+                        wait_time = delay * (2 ** attempt)
+                        logger.info(f"Retrying in {wait_time} seconds...")
+                        await asyncio.sleep(wait_time)
+                        continue
+                    else:
+                        logger.error(f"Database operation failed after {max_retries + 1} attempts")
+                        raise DatabaseConnectionError(
+                            f"Database operation failed after {max_retries + 1} attempts. "
+                            f"Last error: {e}"
+                        ) from e
+                except Exception as e:
+                    # For non-connection errors, don't retry
+                    logger.error(f"Non-recoverable database error: {e}")
+                    raise
+            
+            # This should never be reached, but just in case
+            raise last_exception
+        
+        return wrapper
+    return decorator
+
+async def test_connection() -> bool:
+    """
+    Test database connectivity.
+    
+    Returns:
+        bool: True if connection is successful, False otherwise
+    """
+    try:
+        async with async_session() as session:
+            await session.execute("SELECT 1")
+            return True
+    except Exception as e:
+        logger.error(f"Connection test failed: {e}")
+        return False
+
+async def ensure_connection() -> None:
+    """
+    Ensure database connection is available, attempt to reconnect if needed.
+    
+    Raises:
+        DatabaseConnectionError: If connection cannot be established
+    """
+    if not await test_connection():
+        logger.warning("Database connection lost, attempting to reconnect...")
+        
+        try:
+            # Dispose existing connections and create new ones
+            await async_engine.dispose()
+            
+            # Test connection again
+            if await test_connection():
+                logger.info("Database reconnection successful")
+            else:
+                raise DatabaseConnectionError("Failed to reconnect to database")
+                
+        except Exception as e:
+            raise DatabaseConnectionError(f"Failed to ensure database connection: {e}") from e
+
+async def safe_session_execute(query, *args, **kwargs):
+    """
+    Safely execute a database query with connection retry logic.
+    
+    Args:
+        query: SQL query or statement to execute
+        *args, **kwargs: Additional arguments for session.execute
+        
+    Returns:
+        Query result
+        
+    Raises:
+        DatabaseConnectionError: If operation fails after retries
+    """
+    @retry_db_operation(max_retries=2)
+    async def _execute():
+        async with async_session() as session:
+            try:
+                result = await session.execute(query, *args, **kwargs)
+                return result
+            finally:
+                await session.close()
+    
+    return await _execute()

app/utils/health_utils.py

@@ -0,0 +1,64 @@
+"""
+ANS-specific health check utilities.
+Uses centralized health check from insightfy_utils.telemetry.
+"""
+from insightfy_utils.logging import get_logger
+from insightfy_utils.telemetry.health import HealthCheck, HealthStatus
+from app.sql import database, async_session
+from app.nosql import redis_client
+
+logger = get_logger(__name__)
+
+async def check_postgresql_health() -> tuple[HealthStatus, str]:
+    """Check PostgreSQL database health"""
+    try:
+        async with async_session() as session:
+            await session.execute("SELECT 1")
+        return HealthStatus.HEALTHY, "PostgreSQL is responsive"
+    except Exception as e:
+        logger.error("PostgreSQL health check failed", exc_info=e)
+        return HealthStatus.UNHEALTHY, f"PostgreSQL error: {str(e)}"
+
+async def check_redis_health() -> tuple[HealthStatus, str]:
+    """Check Redis health"""
+    try:
+        await redis_client.ping()
+        return HealthStatus.HEALTHY, "Redis is responsive"
+    except Exception as e:
+        logger.error("Redis health check failed", exc_info=e)
+        return HealthStatus.UNHEALTHY, f"Redis error: {str(e)}"
+
+async def check_mongodb_health() -> tuple[HealthStatus, str]:
+    """Check MongoDB health (if used in ANS)"""
+    try:
+        from app.nosql import mongo_db
+        await mongo_db.command("ping")
+        return HealthStatus.HEALTHY, "MongoDB is responsive"
+    except Exception as e:
+        logger.error("MongoDB health check failed", exc_info=e)
+        return HealthStatus.UNHEALTHY, f"MongoDB error: {str(e)}"
+
+async def perform_comprehensive_health_check() -> dict:
+    """
+    Perform comprehensive health check of all ANS components.
+    Returns health status dictionary.
+    """
+    health = HealthCheck(service_name="ans")
+    
+    # Check PostgreSQL
+    pg_status, pg_message = await check_postgresql_health()
+    health.add_component("postgresql", pg_status, pg_message)
+    
+    # Check Redis
+    redis_status, redis_message = await check_redis_health()
+    health.add_component("redis", redis_status, redis_message)
+    
+    # Check MongoDB (if applicable)
+    try:
+        mongo_status, mongo_message = await check_mongodb_health()
+        health.add_component("mongodb", mongo_status, mongo_message)
+    except ImportError:
+        # MongoDB not used in this service
+        pass
+    
+    return health.get_health()

app/{util → utils}/jwt.py

@@ -1,16 +1,24 @@
 from jose import ExpiredSignatureError, jwt, JWTError
 from settings import SECRET_KEY, ALGORITHM
 
- 
+
 def decode_jwt_token(token: str) -> dict:
+    """
+    Decode and validate JWT token.
+
+    Args:
+        token: JWT token string
+
+    Returns:
+        Decoded payload dict
+
+    Raises:
+        ValueError: If token is expired or invalid
+    """
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        print("payload = ",payload)
         return payload
-
     except ExpiredSignatureError:
         raise ValueError("Token has expired")
-
     except JWTError as e:
-        print(f"❌ Invalid token: {e}")
-        raise ValueError("Invalid token")
+        raise ValueError(f"Invalid token: {e}")

requirements.txt

@@ -1,8 +1,30 @@
-fastapi>=0.95,<1.0
-uvicorn[standard]>=0.22.0
-sqlalchemy 
-psycopg2-binary
-asyncpg
-python-jose==3.4.0
-motor>=3.0
-redis
+# FastAPI and ASGI server
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+pydantic>=2.0.0
+pydantic-settings>=2.0.0
+
+# Database - Async PostgreSQL
+databases>=0.8.0
+sqlalchemy>=2.0.0
+asyncpg>=0.29.0
+
+# Database - MongoDB
+motor>=3.3.0
+
+# Database - Redis
+redis>=5.0.0
+
+# Authentication
+python-jose[cryptography]>=3.3.0
+passlib[bcrypt]>=1.7.4
+python-multipart>=0.0.6
+
+# Utilities
+python-dotenv>=1.0.0
+python-dateutil>=2.8.0
+httpx>=0.24.0
+pandas>=1.5.0
+
+# Centralized utilities (our shared package)
+#insightfy_utils>=0.1.0

settings.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
-
-import os
 from urllib.parse import quote_plus, urlencode
-from dotenv import load_dotenv
+import os
+from insightfy_utils.config import load_env
 
-load_dotenv()  # keep this so local .env is picked up
+load_env()
 
 # ────────────────────────────────────────────────────────────────────────────────
 # Security
@@ -12,83 +11,35 @@ load_dotenv()  # keep this so local .env is picked up
 SECRET_KEY = os.getenv("SECRET_KEY")
 ALGORITHM = os.getenv("ALGORITHM", "HS256")
 
-#postgres keys 
-
-DB_PROTOCOL = os.getenv("DB_PROTOCOL") 
-DB_USER = os.getenv("DB_USER") 
-DB_PASSWORD = os.getenv("DB_PASSWORD")  
-DB_HOST = os.getenv("DB_HOST") 
-DB_PORT = os.getenv("DB_PORT") 
-DB_NAME = os.getenv("DB_NAME") 
-DB_SSLMODE = os.getenv("DB_SSLMODE") 
-
-# settings.py (append these near DATABASE_URI or at the end)
-
-# MongoDB
-MONGO_URI = os.getenv("MONGO_URI")  # e.g., mongodb+srv://user:pass@cluster/db?retryWrites=true&w=majority
-MONGO_DB_NAME = os.getenv("MONGO_DB_NAME", "insightfy-bloom")
-
-# Redis/Cache
-# Prefer a single URL if available (works with redis:// and rediss://)
-CACHE_URL = os.getenv("CACHE_URL") or os.getenv("REDIS_URL")
-
-# Backwards-compat: some setups provide host:port + password separately
-CACHE_URI = os.getenv("CACHE_URI")  # e.g., "localhost:6379"
-CACHE_PASSWORD = os.getenv("CACHE_K")  # password only (optional)
-
-# Optional: default Redis DB index if using host:port form
-CACHE_DB = int(os.getenv("CACHE_DB", "0"))
-
-
 # ────────────────────────────────────────────────────────────────────────────────
-# Database URI builder (split-secret approach)
+# Database - PostgreSQL
 # ────────────────────────────────────────────────────────────────────────────────
-class ConfigError(RuntimeError):
-    pass
-
-
-def _getenv(name: str, *, required: bool = False, default: str | None = None) -> str | None:
-    value = os.getenv(name, default)
-    if required and (value is None or value == ""):
-        raise ConfigError(f"Missing required env var: {name}")
-    return value
-
-
-def build_database_uri(env: dict[str, str] = os.environ) -> str:
-    """
-    Assemble a SQLAlchemy/asyncpg URI from separate env vars.
-    Supports:
-      - URL-encoding for user/password
-      - Default schema via ?options=-csearch_path=<schema>
-      - sslmode=require (e.g., Neon)
-    Honors DATABASE_URI if set (escape hatch/override).
-    """
-
-    # If already provided as a single secret, honor it and return immediately.
-    direct_uri = env.get("DATABASE_URI") or env.get("DATABASE_URL")
-    if direct_uri:
-        return direct_uri
-
-    protocol = _getenv("DB_PROTOCOL", default="postgresql+asyncpg") or "postgresql+asyncpg"
-    user     = _getenv("DB_USER", required=True)
-    password = _getenv("DB_PASSWORD", required=True)
-    host     = _getenv("DB_HOST", default="localhost") or "localhost"
-    port     = _getenv("DB_PORT", default="5432") or "5432"
-    dbname   = _getenv("DB_NAME", required=True)
-
-    # Optional extras
-    schema  = _getenv("DB_SCHEMA")      # e.g., "trans"
-    # sslmode is not supported in asyncpg URI; SSL must be set programmatically if needed
-
-    # No query parameters needed for asyncpg URI
-    q = ""
-
-    # URL-encode credentials so special characters don't break the URI
-    user_enc     = quote_plus(user)
-    password_enc = quote_plus(password)
-
-    return f"{protocol}://{user_enc}:{password_enc}@{host}:{port}/{dbname}{q}"
+DB_PROTOCOL = os.getenv("DB_PROTOCOL", "postgresql+asyncpg")
+DB_USER = os.getenv("DB_USER")
+DB_PASSWORD = os.getenv("DB_PASSWORD")
+DB_HOST = os.getenv("DB_HOST")
+DB_PORT = os.getenv("DB_PORT", "5432")
+DB_NAME = os.getenv("DB_NAME")
+DB_SCHEMA = os.getenv("DB_SCHEMA", "analytics")
+DB_SSLMODE = os.getenv("DB_SSLMODE", "require")
+
+# Build DATABASE_URI
+if all([DB_USER, DB_PASSWORD, DB_HOST, DB_NAME]):
+    DATABASE_URI = f"{DB_PROTOCOL}://{DB_USER}:{quote_plus(DB_PASSWORD)}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
+else:
+    DATABASE_URI = None
 
+# ────────────────────────────────────────────────────────────────────────────────
+# Database - MongoDB
+# ────────────────────────────────────────────────────────────────────────────────
+MONGO_URI = os.getenv('MONGO_URI')
+MONGO_DB_NAME = "insightfy-bloom"
 
-# Public value your app imports
-DATABASE_URI = build_database_uri()
+# ────────────────────────────────────────────────────────────────────────────────
+# Database - Redis/Cache
+# ────────────────────────────────────────────────────────────────────────────────
+CACHE_URI = os.getenv('CACHE_URI')
+CACHE_K = os.getenv('CACHE_K')
+CACHE_URL = os.getenv("CACHE_URL") or os.getenv("REDIS_URL")
+CACHE_PASSWORD = CACHE_K
+CACHE_DB = int(os.getenv("CACHE_DB", "0"))