Deployment via uv

lib/events.py

@@ -2,6 +2,7 @@ from confluent_kafka import Producer
 import json
 import logging
 import os
+from utils.dapr_utils import dapr_http_fallback
 
 logger = logging.getLogger(__name__)
 
@@ -14,7 +15,7 @@ def delivery_report(err, msg):
 
 def publish_task_event(event_type: str, task_data: dict) -> bool:
     """
-    Publish a task event to Kafka.
+    Publish a task event to Dapr pub/sub, with Kafka fallback if Dapr not available.
 
     Args:
         event_type: Type of event (e.g., 'task_created', 'task_completed', 'task_updated')
@@ -23,6 +24,26 @@ def publish_task_event(event_type: str, task_data: dict) -> bool:
     Returns:
         bool: True if event published successfully, False otherwise
     """
+    # First, try to use Dapr sidecar if available
+    dapr_response = dapr_http_fallback(
+        endpoint="/v1.0/publish/task-pubsub/task-events",
+        method="POST",
+        data={
+            "event_type": event_type,
+            "task_data": task_data,
+            "timestamp": task_data.get('updated_at', task_data.get('created_at'))
+        }
+    )
+
+    if dapr_response is not None:
+        # Dapr succeeded
+        logger.info(f"Published {event_type} event via Dapr for task: {task_data.get('id', 'unknown')}")
+        return True
+    else:
+        # Dapr not available, fall back to Kafka
+        logger.info("Dapr sidecar not available, falling back to Kafka")
+
+    # Fallback to Kafka
     try:
         # Get Kafka configuration from environment variables
         bootstrap_servers = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
@@ -60,7 +81,7 @@ def publish_task_event(event_type: str, task_data: dict) -> bool:
         # callbacks to be triggered
         producer.flush()
 
-        logger.info(f"Published {event_type} event for task: {task_data.get('id', 'unknown')}")
+        logger.info(f"Published {event_type} event via Kafka for task: {task_data.get('id', 'unknown')}")
         return True
 
     except Exception as e:

main.py

@@ -1,27 +1,33 @@
 import os
-import logging
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from routes import tasks, chat, chatkit, notifications
 from mcp_server.mcp_server import mcp
 from database import create_db_and_tables
 from dotenv import load_dotenv
-import asyncio
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 from reminder_service import reminder_service
 from recurring_service import recurring_task_service
 from lib.consumer import run_consumer_in_thread
+from utils.logging_config import setup_logging, get_logger
+from utils.monitoring import start_monitoring_server
+from prometheus_client import make_asgi_app
 
 # Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
+setup_logging()
+logger = get_logger(__name__)
 
 # Load environment variables
 load_dotenv()
 
+# Create FastAPI app with monitoring
 app = FastAPI(title="Todo API on Hugging Face")
 
+# Add Prometheus metrics endpoint
+metrics_app = make_asgi_app()
+app.mount("/metrics", metrics_app)
+
 # Initialize scheduler
 scheduler = AsyncIOScheduler()
 
@@ -72,8 +78,14 @@ def startup():
         # Start the Kafka consumer in a background thread
         run_consumer_in_thread()
         logger.info("Kafka consumer started in background thread")
+
+        # Start monitoring server in a background thread
+        import threading
+        monitoring_thread = threading.Thread(target=start_monitoring_server, args=(8001,), daemon=True)
+        monitoring_thread.start()
+        logger.info("Monitoring server started on port 8001")
     except Exception as e:
-        logger.error(f"Error creating database tables: {e}")
+        logger.error(f"Error during startup: {e}")
         raise
 
 @app.on_event("shutdown")

mcp_server/mcp_server.py

@@ -15,11 +15,13 @@ from schemas.input_output_validation import (
     validate_output_format
 )
 from lib.events import publish_task_event
+from utils.monitoring import TASK_CREATED_COUNTER, TASK_COMPLETED_COUNTER, TASK_ERRORS_COUNTER, monitor_task_event
 
 # Initialize the official FastMCP server
 mcp = FastMCP("Focus Task Manager")
 
 @mcp.tool()
+@monitor_task_event("task_created")
 def add_task(user_id: str, title: str, description: Optional[str] = None, priority: Optional[str] = "medium", tags: Optional[List[str]] = None, due_date: Optional[str] = None, is_recurring: Optional[bool] = None, recurrence_pattern: Optional[str] = None) -> str:
     """
     Create a new task in the database.
@@ -85,8 +87,14 @@ def add_task(user_id: str, title: str, description: Optional[str] = None, priori
 
         publish_task_event("task_created", task_data)
 
+        # Increment counter for created tasks
+        TASK_CREATED_COUNTER.inc()
+
         result = f"Success: Created task '{new_task.title}' with ID {new_task.id}"
         return validate_output_format(result, "add_task")
+    except Exception as e:
+        TASK_ERRORS_COUNTER.inc()
+        raise
     finally:
         session.close()
         next(session_gen, None)
@@ -124,6 +132,7 @@ def list_tasks(user_id: str, status: str = "all") -> str:
         next(session_gen, None)
 
 @mcp.tool()
+@monitor_task_event("task_completed")
 def complete_task(user_id: str, task_id: int) -> str:
     """
     Mark a specific task as completed.
@@ -167,9 +176,15 @@ def complete_task(user_id: str, task_id: int) -> str:
 
             publish_task_event("task_completed", task_data)
 
+            # Increment counter for completed tasks
+            TASK_COMPLETED_COUNTER.inc()
+
             result = f"Success: Task {validated_inputs.task_id} marked as completed."
 
         return validate_output_format(result, "complete_task")
+    except Exception as e:
+        TASK_ERRORS_COUNTER.inc()
+        raise
     finally:
         session.close()
         next(session_gen, None)

requirements.txt

@@ -14,4 +14,7 @@ openai-agents
 python-dateutil
 APScheduler
 pywebpush
-confluent-kafka
+confluent-kafka
+dapr
+dapr-ext-fastapi
+python-json-logger

utils/dapr_utils.py

@@ -0,0 +1,65 @@
+"""
+Dapr utility functions for cloud-ready applications.
+Provides graceful fallback when Dapr sidecar is not available.
+"""
+import os
+import logging
+import requests
+import json
+
+logger = logging.getLogger(__name__)
+
+def dapr_http_fallback(endpoint: str, method: str = "POST", data=None, headers=None):
+    """
+    Generic Dapr HTTP fallback function.
+    Tries to use Dapr sidecar, falls back gracefully if not available.
+
+    Args:
+        endpoint: Dapr endpoint (e.g., "/v1.0/publish/pubsub_name/topic_name")
+        method: HTTP method (GET, POST, PUT, DELETE)
+        data: Data to send in request body
+        headers: Additional headers to send
+
+    Returns:
+        Response from Dapr or fallback, or None if both fail
+    """
+    dapr_port = os.getenv("DAPR_HTTP_PORT", "3500")
+    dapr_url = f"http://localhost:{dapr_port}{endpoint}"
+
+    # Check if Dapr sidecar is available
+    try:
+        dapr_health_url = f"http://localhost:{dapr_port}/v1.0/healthz"
+        health_response = requests.get(dapr_health_url, timeout=2)
+
+        if health_response.status_code == 200:
+            # Dapr sidecar is available, use it
+            req_headers = headers or {}
+            req_headers["Content-Type"] = "application/json"
+
+            if method.upper() == "POST":
+                response = requests.post(dapr_url, json=data, headers=req_headers, timeout=10)
+            elif method.upper() == "GET":
+                response = requests.get(dapr_url, headers=req_headers, timeout=10)
+            elif method.upper() == "PUT":
+                response = requests.put(dapr_url, json=data, headers=req_headers, timeout=10)
+            elif method.upper() == "DELETE":
+                response = requests.delete(dapr_url, headers=req_headers, timeout=10)
+            else:
+                logger.error(f"Unsupported HTTP method: {method}")
+                return None
+
+            if response.status_code in [200, 204]:
+                logger.debug(f"Dapr call successful: {method} {endpoint}")
+                return response
+            else:
+                logger.warning(f"Dapr call failed with status {response.status_code}, endpoint: {endpoint}")
+                return None
+        else:
+            logger.warning(f"Dapr sidecar not available, skipping Dapr call to {endpoint}")
+            return None
+    except requests.exceptions.RequestException:
+        logger.warning(f"Dapr sidecar not available, skipping Dapr call to {endpoint}")
+        return None
+    except Exception as e:
+        logger.warning(f"Dapr unavailable for {endpoint}: {str(e)}")
+        return None

utils/logging_config.py

@@ -0,0 +1,44 @@
+import logging
+import sys
+from pythonjsonlogger import jsonlogger
+import os
+from datetime import datetime
+
+def setup_logging():
+    """Setup structured logging for the application"""
+
+    # Get log level from environment, default to INFO
+    log_level = os.getenv('LOG_LEVEL', 'INFO').upper()
+
+    # Create a custom JSON formatter
+    json_formatter = jsonlogger.JsonFormatter(
+        '%(asctime)s %(name)s %(levelname)s %(filename)s %(lineno)d %(message)s',
+        datefmt='%Y-%m-%dT%H:%M:%S'
+    )
+
+    # Setup root logger
+    root_logger = logging.getLogger()
+    root_logger.setLevel(getattr(logging, log_level))
+
+    # Clear any existing handlers
+    root_logger.handlers.clear()
+
+    # Create handler for stdout
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(json_formatter)
+    root_logger.addHandler(handler)
+
+    # Also setup specific loggers for different components
+    logging.getLogger('uvicorn').setLevel(getattr(logging, log_level))
+    logging.getLogger('uvicorn.access').setLevel(getattr(logging, log_level))
+    logging.getLogger('uvicorn.error').setLevel(getattr(logging, log_level))
+    logging.getLogger('fastapi').setLevel(getattr(logging, log_level))
+    logging.getLogger('sqlalchemy').setLevel(logging.WARNING)  # Reduce SQLAlchemy noise
+    logging.getLogger('confluent_kafka').setLevel(getattr(logging, log_level))
+
+def get_logger(name: str) -> logging.Logger:
+    """Get a logger instance with the specified name"""
+    return logging.getLogger(name)
+
+# Initialize logging when module is imported
+setup_logging()

utils/monitoring.py

@@ -0,0 +1,57 @@
+from prometheus_client import Counter, Histogram, Gauge, start_http_server
+import time
+import logging
+from functools import wraps
+
+# Create loggers
+logger = logging.getLogger(__name__)
+
+# Define metrics
+REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status_code'])
+REQUEST_DURATION = Histogram('http_request_duration_seconds', 'Duration of HTTP requests in seconds', ['method', 'endpoint'])
+ACTIVE_TASKS = Gauge('active_tasks_count', 'Number of active tasks')
+TASK_EVENTS_PROCESSED = Counter('task_events_processed_total', 'Total task events processed', ['event_type'])
+
+class MonitoringMiddleware:
+    """Custom middleware to collect metrics for FastAPI application"""
+
+    def __init__(self):
+        self.logger = logging.getLogger(self.__class__.__name__)
+
+    def record_request(self, method: str, endpoint: str, status_code: int, duration: float):
+        """Record request metrics"""
+        REQUEST_COUNT.labels(method=method, endpoint=endpoint, status_code=status_code).inc()
+        REQUEST_DURATION.labels(method=method, endpoint=endpoint).observe(duration)
+
+        self.logger.info(f"Request metrics recorded: {method} {endpoint} {status_code} {duration}s")
+
+def monitor_task_event(event_type: str):
+    """Decorator to monitor task event processing"""
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            start_time = time.time()
+            try:
+                result = func(*args, **kwargs)
+                TASK_EVENTS_PROCESSED.labels(event_type=event_type).inc()
+                duration = time.time() - start_time
+                logger.info(f"Task event {event_type} processed in {duration:.2f}s")
+                return result
+            except Exception as e:
+                logger.error(f"Error processing task event {event_type}: {str(e)}")
+                raise
+        return wrapper
+    return decorator
+
+def start_monitoring_server(port: int = 8001):
+    """Start the Prometheus metrics server"""
+    try:
+        start_http_server(port)
+        logger.info(f"Monitoring server started on port {port}")
+    except Exception as e:
+        logger.error(f"Failed to start monitoring server: {str(e)}")
+
+# Predefined metrics for common operations
+TASK_CREATED_COUNTER = Counter('tasks_created_total', 'Total tasks created')
+TASK_COMPLETED_COUNTER = Counter('tasks_completed_total', 'Total tasks completed')
+TASK_ERRORS_COUNTER = Counter('task_errors_total', 'Total task-related errors')