labFusion/services/service-adapters/CLEAN_CODE.md

Service Adapters Clean Code Implementation

This document outlines the clean code principles and best practices implemented in the LabFusion Service Adapters service (Python FastAPI).

🏗️ Architecture Overview

The Service Adapters follow a modular architecture with clear separation of concerns:

service-adapters/
├── main.py              # FastAPI application entry point
├── models/              # Pydantic schemas (Domain Layer)
│   ├── __init__.py
│   └── schemas.py
├── routes/              # API endpoints (Presentation Layer)
│   ├── __init__.py
│   ├── general.py       # General endpoints
│   ├── home_assistant.py # Home Assistant integration
│   ├── frigate.py       # Frigate integration
│   ├── immich.py        # Immich integration
│   └── events.py        # Event management
├── services/            # Business logic (Service Layer)
│   ├── __init__.py
│   ├── config.py        # Configuration management
│   └── redis_client.py  # Redis connection
├── requirements.txt     # Dependencies
└── README.md           # Service documentation

🧹 Clean Code Principles Applied

1. Single Responsibility Principle (SRP)

Route Modules

• general.py: Only handles general endpoints (health, services, root)
• home_assistant.py: Only manages Home Assistant integration
• frigate.py: Only handles Frigate camera system integration
• immich.py: Only manages Immich photo management integration
• events.py: Only handles event publishing and retrieval

Service Modules

• config.py: Only manages service configurations
• redis_client.py: Only handles Redis connections and operations

Model Modules

• schemas.py: Only contains Pydantic data models

2. Open/Closed Principle (OCP)

Extensible Route Design

# Easy to add new service integrations
from fastapi import APIRouter

router = APIRouter(prefix="/home-assistant", tags=["Home Assistant"])

# New integrations can be added without modifying existing code
# Just create new route files and include them in main.py

Configurable Services

# services/config.py
SERVICES = {
    "home_assistant": {
        "name": "Home Assistant",
        "url": os.getenv("HOME_ASSISTANT_URL", "http://homeassistant.local:8123"),
        "token": os.getenv("HOME_ASSISTANT_TOKEN"),
        "enabled": True
    }
    # Easy to add new services
}

3. Dependency Inversion Principle (DIP)

Dependency Injection

# main.py
from services.redis_client import get_redis_client

app.include_router(general_router, dependencies=[Depends(get_redis_client)])
app.include_router(home_assistant_router, dependencies=[Depends(get_redis_client)])

Interface-Based Design

# services/redis_client.py
class RedisClient:
    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        self.client = None
    
    async def connect(self):
        # Implementation details hidden
        pass

4. Interface Segregation Principle (ISP)

Focused Route Groups

• Each route module only exposes endpoints relevant to its service
• Clear API boundaries
• Minimal dependencies between modules

📝 Code Quality Improvements

1. Naming Conventions

Clear, Descriptive Names

# Good: Clear purpose
class ServiceStatus(BaseModel):
    name: str
    status: str
    response_time: Optional[str] = None

# Good: Descriptive function names
async def get_home_assistant_entities()
async def get_frigate_events()
async def publish_event(event_data: EventData)

Consistent Naming

• Classes: PascalCase (e.g., ServiceStatus, EventData)
• Functions: snake_case (e.g., get_home_assistant_entities)
• Variables: snake_case (e.g., service_config)
• Constants: UPPER_SNAKE_CASE (e.g., SERVICES)

2. Function Design

Small, Focused Functions

@router.get("/entities")
async def get_home_assistant_entities():
    """Get all Home Assistant entities."""
    try:
        entities = await fetch_ha_entities()
        return {"entities": entities}
    except Exception as e:
        logger.error(f"Error fetching HA entities: {e}")
        raise HTTPException(status_code=500, detail="Failed to fetch entities")

Single Level of Abstraction

• Route functions handle HTTP concerns only
• Business logic delegated to service functions
• Data validation handled by Pydantic models

3. Error Handling

Consistent Error Responses

try:
    result = await some_operation()
    return result
except ConnectionError as e:
    logger.error(f"Connection error: {e}")
    raise HTTPException(status_code=503, detail="Service unavailable")
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    raise HTTPException(status_code=500, detail="Internal server error")

Proper HTTP Status Codes

• 200: Success
• 404: Not found
• 503: Service unavailable
• 500: Internal server error

4. Data Validation

Pydantic Models

class EventData(BaseModel):
    event_type: str
    service: str
    timestamp: datetime
    data: Dict[str, Any]
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

Input Validation

@router.post("/publish-event")
async def publish_event(event_data: EventData):
    # Pydantic automatically validates input
    await redis_client.publish_event(event_data)
    return {"message": "Event published successfully"}

🔧 FastAPI Best Practices

1. Router Organization

Modular Route Structure

# routes/home_assistant.py
from fastapi import APIRouter, HTTPException, Depends
from models.schemas import HAAttributes, HAEntity
from services.redis_client import RedisClient

router = APIRouter(prefix="/home-assistant", tags=["Home Assistant"])

@router.get("/entities")
async def get_entities(redis: RedisClient = Depends(get_redis_client)):
    # Implementation

Clear API Documentation

@router.get(
    "/entities",
    response_model=Dict[str, List[HAEntity]],
    summary="Get Home Assistant entities",
    description="Retrieve all entities from Home Assistant"
)
async def get_home_assistant_entities():
    # Implementation

2. Dependency Injection

Redis Client Injection

# services/redis_client.py
async def get_redis_client() -> RedisClient:
    redis = RedisClient(REDIS_URL)
    await redis.connect()
    return redis

Service Configuration

# services/config.py
def get_service_config(service_name: str) -> Dict[str, Any]:
    return SERVICES.get(service_name, {})

3. OpenAPI Documentation

Comprehensive API Documentation

app = FastAPI(
    title="LabFusion Service Adapters",
    description="Integration adapters for homelab services",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

Detailed Endpoint Documentation

@router.get(
    "/events",
    response_model=Dict[str, List[FrigateEvent]],
    summary="Get Frigate events",
    description="Retrieve recent events from Frigate camera system",
    responses={
        200: {"description": "Successfully retrieved events"},
        503: {"description": "Frigate service unavailable"}
    }
)

📊 Data Layer Best Practices

1. Pydantic Model Design

Clean Model Structure

class ServiceStatus(BaseModel):
    name: str
    status: str
    response_time: Optional[str] = None
    last_check: Optional[datetime] = None
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

Proper Type Hints

from typing import List, Dict, Optional, Any
from datetime import datetime

class EventData(BaseModel):
    event_type: str
    service: str
    timestamp: datetime
    data: Dict[str, Any]

2. Configuration Management

Environment-Based Configuration

# services/config.py
import os

SERVICES = {
    "home_assistant": {
        "name": "Home Assistant",
        "url": os.getenv("HOME_ASSISTANT_URL", "http://homeassistant.local:8123"),
        "token": os.getenv("HOME_ASSISTANT_TOKEN"),
        "enabled": os.getenv("HOME_ASSISTANT_ENABLED", "true").lower() == "true"
    }
}

Centralized Configuration

• All service configurations in one place
• Environment variable support
• Default values for development

🚀 Performance Optimizations

1. Async/Await Usage

async def fetch_ha_entities() -> List[HAEntity]:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{HA_URL}/api/states")
        return response.json()

2. Connection Pooling

# services/redis_client.py
class RedisClient:
    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        self.client = None
    
    async def connect(self):
        self.client = await aioredis.from_url(self.redis_url)

3. Error Handling

async def safe_api_call(url: str, headers: Dict[str, str]) -> Optional[Dict]:
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url, headers=headers, timeout=5.0)
            response.raise_for_status()
            return response.json()
    except httpx.TimeoutException:
        logger.warning(f"Timeout calling {url}")
        return None
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP error calling {url}: {e}")
        return None

🧪 Testing Strategy

1. Unit Testing

import pytest
from unittest.mock import AsyncMock, patch
from services.config import SERVICES

@pytest.mark.asyncio
async def test_get_services():
    with patch('services.redis_client.get_redis_client') as mock_redis:
        # Test implementation
        pass

2. Integration Testing

@pytest.mark.asyncio
async def test_home_assistant_integration():
    # Test actual HA API calls
    pass

3. Test Structure

# tests/test_home_assistant.py
import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_get_ha_entities():
    response = client.get("/home-assistant/entities")
    assert response.status_code == 200

📋 Code Review Checklist

Route Layer

• Single responsibility per route module
• Proper HTTP status codes
• Input validation with Pydantic
• Error handling
• OpenAPI documentation

Service Layer

• Business logic only
• No direct HTTP calls in business logic
• Proper exception handling
• Async/await usage
• Clear function names

Model Layer

• Proper Pydantic models
• Type hints
• Validation constraints
• JSON serialization

Configuration

• Environment variable support
• Default values
• Centralized configuration
• Service-specific settings

🎯 Benefits Achieved

1. Maintainability

• Clear separation of concerns
• Easy to modify and extend
• Consistent patterns throughout
• Self-documenting code

2. Testability

• Isolated modules
• Mockable dependencies
• Clear interfaces
• Testable business logic

3. Performance

• Async/await for non-blocking operations
• Connection pooling
• Efficient error handling
• Resource management

4. Scalability

• Modular architecture
• Easy to add new service integrations
• Horizontal scaling support
• Configuration-driven services

🔮 Future Improvements

Potential Enhancements

1. Circuit Breaker Pattern: Fault tolerance
2. Retry Logic: Automatic retry with backoff
3. Caching: Redis caching for frequently accessed data
4. Metrics: Prometheus metrics integration
5. Health Checks: Comprehensive health monitoring

Monitoring & Observability

1. Logging: Structured logging with correlation IDs
2. Tracing: Distributed tracing
3. Metrics: Service performance metrics
4. Alerts: Service availability alerts

This clean code implementation makes the Service Adapters more maintainable, testable, and scalable while following FastAPI and Python best practices.