implement n8n client

application/use_cases/container.py

@@ -29,11 +29,12 @@ class UseCaseContainer:
     def _init_use_cases(self):
         """Initialize all use cases with dependencies."""
         
-        # Process job use case (needed by async extractor)
+        # Process job use case (updated with notification service)
         self.process_job = ProcessJobUseCase(
             job_repository=self.services.job_repository,
             ffmpeg_service=self.services.ffmpeg_service,
-            file_repository=self.services.file_repository
+            file_repository=self.services.file_repository,
+            notification_service=self.services.notification_service  # Added notification service
         )
         
         # Async extraction use case

application/use_cases/process_job.py

@@ -36,13 +36,22 @@ class FileRepository(Protocol):
     async def delete_file(self, file_path: str) -> bool:
         ...
 
+class NotificationService(Protocol):
+    """Protocol for notification service."""
+    async def send_job_completion_notification(self, 
+                                             job_id: str, 
+                                             status: str,
+                                             processing_time: float) -> Any:
+        ...
+
 class ProcessJobUseCase:
     """Use case for processing a queued extraction job."""
     
-    def __init__(self, file_repository, ffmpeg_service, job_repository):
+    def __init__(self, file_repository, ffmpeg_service, job_repository, notification_service):
         self.file_repository = file_repository
         self.ffmpeg_service = ffmpeg_service
         self.job_repository = job_repository
+        self.notification_service = notification_service
         self.file_processor = LocalFileProcessor(file_repository)
     
     async def execute(self, job_id: str, request: ExtractionRequestDTO):
@@ -78,6 +87,12 @@ class ProcessJobUseCase:
                 output_path=output_key,
                 processing_time=processing_time
             )
+
+            await self.notification_service.send_job_completion_notification(
+                job_id=job_id,
+                status="completed",
+                processing_time=processing_time
+            )
             
             logger.info(f"Job {job_id} completed in {processing_time:.2f} seconds")
             

domain/services/notification_service.py

@@ -0,0 +1,19 @@
+from typing import Protocol
+from dataclasses import dataclass
+
+@dataclass
+class NotificationRequest:
+    message: str
+
+@dataclass 
+class NotificationResponse:
+    acknowledged: bool
+
+class NotificationService(Protocol):
+    """Protocol for sending notifications after job completion."""
+    
+    async def send_job_completion_notification(self, 
+                                             job_id: str, 
+                                             status: str,
+                                             processing_time: float) -> NotificationResponse:
+        ...

infrastructure/clients/n8n/exceptions.py

@@ -0,0 +1,17 @@
+class APIClientError(Exception):
+    """Base exception for API client errors."""
+    pass
+
+
+class APIConnectionError(APIClientError):
+    """Raised when connection to API fails."""
+    pass
+
+
+class APIResponseError(APIClientError):
+    """Raised when API returns an error response."""
+    
+    def __init__(self, message: str, status_code: int, response_body: str = ""):
+        super().__init__(message)
+        self.status_code = status_code
+        self.response_body = response_body

infrastructure/clients/n8n/models.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+from typing import Protocol
+
+
+@dataclass
+class WebhooksRequest:
+    """Request model for webhooks endpoint."""
+    message: str
+    job_id: str
+
+
+@dataclass
+class WebhooksResponse:
+    """Response model for webhooks endpoint."""
+    acknowledged: bool
+
+
+class N8NClientProtocol(Protocol):
+    """Protocol defining the API client interface."""
+    
+    async def post_completion_event(self, data: WebhooksRequest) -> WebhooksResponse:
+        """Post to webhooks endpoint."""
+        ...

infrastructure/clients/n8n/n8n_client.py

@@ -0,0 +1,114 @@
+import json
+import logging
+from typing import Dict, Any, Optional
+import httpx
+from .settings import ClientSettings
+from .models import WebhooksRequest, WebhooksResponse, N8NClientProtocol
+from .exceptions import APIClientError, APIConnectionError, APIResponseError
+from dataclasses import asdict
+
+
+class N8NClient(N8NClientProtocol):
+    """HTTP client for API interactions with automatic auth and logging."""
+    
+    def __init__(self, settings: ClientSettings, logger: logging.Logger):
+        self.settings = settings
+        self.logger = logger
+        self._client = httpx.AsyncClient(
+            base_url=settings.base_url,
+            timeout=settings.timeout,
+            headers=self._get_default_headers()
+        )
+    
+    def _get_default_headers(self) -> Dict[str, str]:
+        """Get default headers including authentication."""
+        return {
+            "Authorization": f"Bearer {self.settings.token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json"
+        }
+    
+    async def _make_request(
+        self, 
+        method: str, 
+        endpoint: str, 
+        data: Optional[Dict[str, Any]] = None,
+        custom_headers: Optional[Dict[str, str]] = None
+    ) -> Dict[str, Any]:
+        """Make HTTP request with automatic logging and error handling."""
+        url = endpoint
+        headers = dict(self._client.headers)
+
+        if custom_headers:
+            headers.update(custom_headers)
+        
+        # Log request
+        self.logger.info(
+            f"Making {method} request to {url}",
+            extra={
+                "method": method,
+                "url": url,
+                "headers": {k: v if k.lower() != "authorization" else "Bearer ***" for k, v in headers.items()},
+                "payload": data
+            }
+        )
+        
+        try:
+            response = await self._client.request(
+                method=method,
+                url=url,
+                json=data,
+                headers=custom_headers  # Pass custom headers to the request
+            )
+            
+            # Log response
+            self.logger.info(
+                f"Received response from {url}",
+                extra={
+                    "status_code": response.status_code,
+                    "response_headers": dict(response.headers),
+                    "response_body": response.text
+                }
+            )
+            
+            # Handle HTTP errors
+            if response.status_code >= 400:
+                raise APIResponseError(
+                    f"API request failed with status {response.status_code}",
+                    status_code=response.status_code,
+                    response_body=response.text
+                )
+            
+            return response.json()
+            
+        except httpx.RequestError as e:
+            self.logger.error(f"Connection error for {url}: {str(e)}")
+            raise APIConnectionError(f"Failed to connect to API: {str(e)}")
+        except json.JSONDecodeError as e:
+            self.logger.error(f"Failed to parse JSON response from {url}: {str(e)}")
+            raise APIResponseError(f"Invalid JSON response: {str(e)}", response.status_code)
+    
+    async def post_completion_event(self, data: WebhooksRequest) -> WebhooksResponse:
+        """Post to webhooks endpoint."""
+        from dataclasses import asdict
+        payload = asdict(data)
+        
+        # Extract job_id for header and remove from payload
+        job_id = payload.pop("job_id")
+        custom_headers = {"rowID": job_id}
+        
+        response_data = await self._make_request("POST", "/lovable-analysis", payload, custom_headers)
+        
+        return WebhooksResponse(acknowledged=response_data.get("acknowledged", False))
+    
+    async def close(self):
+        """Close the HTTP client."""
+        await self._client.aclose()
+    
+    async def __aenter__(self):
+        """Async context manager entry."""
+        return self
+    
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit."""
+        await self.close()

infrastructure/clients/n8n/settings.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class ClientSettings:
+    """Settings for the HTTP client."""
+    base_url: str
+    token: str
+    timeout: Optional[int] = 30

infrastructure/config/settings.py

@@ -85,6 +85,11 @@ class Settings(BaseSettings):
         'm4a': 'audio/mp4',
         'ogg': 'audio/ogg'
     }
+
+    # N8N Configuration
+    n8n_base_url: str = Field(default="http://localhost:5678", env="N8N_BASE_URL")
+    n8n_token: str = Field(env="N8N_TOKEN")
+    n8n_timeout: int = Field(default=30, env="N8N_TIMEOUT")
     
     class Config:
         env_file = ".env"

infrastructure/services/container.py

@@ -1,6 +1,7 @@
 """Dependency injection container for services."""
 from typing import Optional
 from pathlib import Path
+import logging
 
 from ..config.settings import settings
 from ..repositories.job_repository import InMemoryJobRepository
@@ -8,6 +9,11 @@ from ..providers.file_storage_provider import create_storage_from_settings
 from .ffmpeg_service import FFmpegService
 from .file_cleanup_service import FileCleanupService
 
+# N8N related imports
+from ..clients.n8n.n8n_client import N8NClient
+from ..clients.n8n.settings import ClientSettings
+from .n8n_notification_service import N8NNotificationService
+
 class ServiceContainer:
     """Container for all infrastructure services."""
     
@@ -33,6 +39,9 @@ class ServiceContainer:
             cleanup_interval_seconds=settings.cleanup_interval_seconds,
             retention_hours=settings.file_retention_hours
         )
+        
+        # Create N8N notification service
+        self.notification_service = self._create_notification_service()
     
     def _create_file_repository(self):
         """Create file repository based on settings."""
@@ -52,6 +61,24 @@ class ServiceContainer:
         else:
             raise ValueError(f"Unsupported storage type: {settings.storage_type}")
     
+    def _create_notification_service(self) -> N8NNotificationService:
+        """Create N8N notification service."""
+        # Create N8N client settings
+        n8n_settings = ClientSettings(
+            base_url=settings.n8n_base_url,
+            token=settings.n8n_token,
+            timeout=getattr(settings, 'n8n_timeout', 30)
+        )
+        
+        # Create logger for N8N client
+        logger = logging.getLogger("n8n_client")
+        
+        # Create N8N client
+        n8n_client = N8NClient(n8n_settings, logger)
+        
+        # Create and return notification service
+        return N8NNotificationService(n8n_client)
+    
     @classmethod
     def get_instance(cls) -> 'ServiceContainer':
         """Get singleton instance of service container."""
@@ -66,6 +93,9 @@ class ServiceContainer:
     async def shutdown(self):
         """Cleanup services on shutdown."""
         await self.cleanup_service.stop()
+        # Close N8N client if it has async cleanup
+        if hasattr(self.notification_service.n8n_client, 'close'):
+            await self.notification_service.n8n_client.close()
 
 # Convenience function for getting services
 def get_services() -> ServiceContainer:

infrastructure/services/n8n_notification_service.py

@@ -0,0 +1,29 @@
+from domain.services.notification_service import NotificationService, NotificationRequest, NotificationResponse
+import logging
+
+from infrastructure.clients.n8n.n8n_client import N8NClient
+
+logger = logging.getLogger(__name__)
+
+class N8NNotificationService(NotificationService):
+    """N8N implementation of notification service."""
+    
+    def __init__(self, n8n_client: N8NClient):
+        self.n8n_client = n8n_client
+    
+    async def send_job_completion_notification(self, 
+                                             job_id: str, 
+                                             status: str,
+                                             processing_time: float) -> NotificationResponse:
+        """Send job completion notification via N8N."""
+        try:
+            message = f"Job {job_id} {status} in {processing_time:.2f}s"
+            request = NotificationRequest(message=message, job_id=job_id)
+            
+            response = await self.n8n_client.post_completion_event(request)
+            return NotificationResponse(acknowledged=response.acknowledged)
+            
+        except Exception as e:
+            logger.error(f"Failed to send notification for job {job_id}: {e}")
+            # You might want to return False or re-raise depending on your error handling strategy
+            return NotificationResponse(acknowledged=False)

requirements.txt

@@ -5,6 +5,7 @@ uvicorn[standard]==0.24.0
 python-multipart==0.0.6
 pydantic==2.5.0
 pydantic-settings==2.1.0
+httpx==0.27.2
 
 # File Processing
 aiofiles==23.2.1