Instruction file imported from humanstack/vibe-coding-template (.cursor/rules/templates/service-class-template.mdc). Copyright stays with the author.

Service Class Template

Use this template when creating new service classes:

from typing import Dict, List, Any, Optional, TypeVar, Generic, Type
from abc import ABC, abstractmethod
import logging
from datetime import datetime

from app.core.config import settings

# Set up logging
logger = logging.getLogger(__name__)

T = TypeVar("T")

class BaseService(ABC, Generic[T]):
    """Abstract base class for all services."""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.logger = logging.getLogger(f"{__name__}.{service_name}")

    @abstractmethod
    async def initialize(self) -> None:
        """Initialize the service. Override in subclasses."""
        pass

    @abstractmethod
    async def health_check(self) -> bool:
        """Check if the service is healthy. Override in subclasses."""
        pass

class ExampleService(BaseService[dict]):
    """
    Example service class demonstrating the standard pattern.

    This service handles [describe what the service does].
    """

    def __init__(self):
        super().__init__("ExampleService")
        self._initialized = False
        self._client = None

    async def initialize(self) -> None:
        """Initialize the service with necessary configurations."""
        try:
            self.logger.info("Initializing ExampleService")

            # Initialize external clients, connections, etc.
            self._client = self._create_client()

            # Perform any setup operations
            await self._setup()

            self._initialized = True
            self.logger.info("ExampleService initialized successfully")

        except Exception as e:
            self.logger.error(f"Failed to initialize ExampleService: {str(e)}")
            raise

    async def health_check(self) -> bool:
        """Check if the service is healthy and operational."""
        try:
            if not self._initialized:
                return False

            # Perform health check operations
            # e.g., ping external service, check database connection, etc.
            result = await self._ping_service()

            self.logger.debug(f"Health check result: {result}")
            return result

        except Exception as e:
            self.logger.error(f"Health check failed: {str(e)}")
            return False

    async def create_item(self, data: Dict[str, Any]) -> dict:
        """
        Create a new item.

        Args:
            data: Dictionary containing item data

        Returns:
            Created item data

        Raises:
            ValueError: If data is invalid
            RuntimeError: If service is not initialized or operation fails
        """
        self._ensure_initialized()

        try:
            # Validate input data
            validated_data = self._validate_create_data(data)

            self.logger.info(f"Creating item with data: {validated_data}")

            # Perform the creation operation
            result = await self._perform_create(validated_data)

            self.logger.info(f"Successfully created item: {result.get('id')}")
            return result

        except ValueError as e:
            self.logger.warning(f"Invalid data for item creation: {str(e)}")
            raise
        except Exception as e:
            self.logger.error(f"Failed to create item: {str(e)}")
            raise RuntimeError(f"Item creation failed: {str(e)}")

    async def get_item(self, item_id: str) -> Optional[dict]:
        """
        Retrieve an item by ID.

        Args:
            item_id: Unique identifier for the item

        Returns:
            Item data if found, None otherwise

        Raises:
            ValueError: If item_id is invalid
            RuntimeError: If service is not initialized or operation fails
        """
        self._ensure_initialized()

        if not item_id or not isinstance(item_id, str):
            raise ValueError("Valid item_id is required")

        try:
            self.logger.debug(f"Retrieving item: {item_id}")

            result = await self._perform_get(item_id)

            if result:
                self.logger.debug(f"Found item: {item_id}")
            else:
                self.logger.debug(f"Item not found: {item_id}")

            return result

        except Exception as e:
            self.logger.error(f"Failed to retrieve item {item_id}: {str(e)}")
            raise RuntimeError(f"Item retrieval failed: {str(e)}")

    async def list_items(self, filters: Optional[Dict[str, Any]] = None, limit: int = 100) -> List[dict]:
        """
        List items with optional filtering.

        Args:
            filters: Optional dictionary of filters to apply
            limit: Maximum number of items to return

        Returns:
            List of item data

        Raises:
            ValueError: If parameters are invalid
            RuntimeError: If service is not initialized or operation fails
        """
        self._ensure_initialized()

        if limit <= 0 or limit > 1000:
            raise ValueError("Limit must be between 1 and 1000")

        try:
            self.logger.debug(f"Listing items with filters: {filters}, limit: {limit}")

            result = await self._perform_list(filters or {}, limit)

            self.logger.debug(f"Retrieved {len(result)} items")
            return result

        except Exception as e:
            self.logger.error(f"Failed to list items: {str(e)}")
            raise RuntimeError(f"Item listing failed: {str(e)}")

    async def update_item(self, item_id: str, data: Dict[str, Any]) -> dict:
        """
        Update an existing item.

        Args:
            item_id: Unique identifier for the item
            data: Dictionary containing updated data

        Returns:
            Updated item data

        Raises:
            ValueError: If parameters are invalid
            RuntimeError: If service is not initialized or operation fails
        """
        self._ensure_initialized()

        if not item_id or not isinstance(item_id, str):
            raise ValueError("Valid item_id is required")

        try:
            # Validate update data
            validated_data = self._validate_update_data(data)

            self.logger.info(f"Updating item {item_id} with data: {validated_data}")

            result = await self._perform_update(item_id, validated_data)

            self.logger.info(f"Successfully updated item: {item_id}")
            return result

        except ValueError as e:
            self.logger.warning(f"Invalid data for item update: {str(e)}")
            raise
        except Exception as e:
            self.logger.error(f"Failed to update item {item_id}: {str(e)}")
            raise RuntimeError(f"Item update failed: {str(e)}")

    async def delete_item(self, item_id: str) -> bool:
        """
        Delete an item by ID.

        Args:
            item_id: Unique identifier for the item

        Returns:
            True if deletion was successful

        Raises:
            ValueError: If item_id is invalid
            RuntimeError: If service is not initialized or operation fails
        """
        self._ensure_initialized()

        if not item_id or not isinstance(item_id, str):
            raise ValueError("Valid item_id is required")

        try:
            self.logger.info(f"Deleting item: {item_id}")

            success = await self._perform_delete(item_id)

            if success:
                self.logger.info(f"Successfully deleted item: {item_id}")
            else:
                self.logger.warning(f"Item not found for deletion: {item_id}")

            return success

        except Exception as e:
            self.logger.error(f"Failed to delete item {item_id}: {str(e)}")
            raise RuntimeError(f"Item deletion failed: {str(e)}")

    # Private helper methods

    def _ensure_initialized(self) -> None:
        """Ensure the service is initialized before operations."""
        if not self._initialized:
            raise RuntimeError(f"{self.service_name} is not initialized")

    def _create_client(self):
        """Create and configure the external client."""
        # Implement client creation logic
        # e.g., return ExternalClient(api_key=settings.API_KEY)
        pass

    async def _setup(self) -> None:
        """Perform any additional setup operations."""
        # Implement setup logic
        pass

    async def _ping_service(self) -> bool:
        """Ping the external service to check connectivity."""
        # Implement ping logic
        return True

    def _validate_create_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Validate data for item creation."""
        required_fields = ["name"]  # Define required fields

        for field in required_fields:
            if field not in data:
                raise ValueError(f"Missing required field: {field}")

        # Add timestamp
        data["created_at"] = datetime.utcnow().isoformat()

        return data

    def _validate_update_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Validate data for item update."""
        # Remove None values
        validated_data = {k: v for k, v in data.items() if v is not None}

        if not validated_data:
            raise ValueError("No valid fields to update")

        # Add timestamp
        validated_data["updated_at"] = datetime.utcnow().isoformat()

        return validated_data

    async def _perform_create(self, data: Dict[str, Any]) -> dict:
        """Perform the actual creation operation."""
        # Implement creation logic
        # e.g., return await self._client.create(data)
        return {"id": "example_id", **data}

    async def _perform_get(self, item_id: str) -> Optional[dict]:
        """Perform the actual retrieval operation."""
        # Implement retrieval logic
        # e.g., return await self._client.get(item_id)
        return None

    async def _perform_list(self, filters: Dict[str, Any], limit: int) -> List[dict]:
        """Perform the actual listing operation."""
        # Implement listing logic
        # e.g., return await self._client.list(filters, limit)
        return []

    async def _perform_update(self, item_id: str, data: Dict[str, Any]) -> dict:
        """Perform the actual update operation."""
        # Implement update logic
        # e.g., return await self._client.update(item_id, data)
        return {"id": item_id, **data}

    async def _perform_delete(self, item_id: str) -> bool:
        """Perform the actual deletion operation."""
        # Implement deletion logic
        # e.g., return await self._client.delete(item_id)
        return True

# Service factory pattern (optional)
class ServiceFactory:
    """Factory for creating service instances."""

    _instances: Dict[str, Any] = {}

    @classmethod
    async def get_service(cls, service_type: str) -> Any:
        """Get or create a service instance."""
        if service_type not in cls._instances:
            if service_type == "example":
                service = ExampleService()
                await service.initialize()
                cls._instances[service_type] = service
            else:
                raise ValueError(f"Unknown service type: {service_type}")

        return cls._instances[service_type]

    @classmethod
    async def health_check_all(cls) -> Dict[str, bool]:
        """Check health of all initialized services."""
        results = {}
        for service_type, service in cls._instances.items():
            results[service_type] = await service.health_check()
        return results

Key Template Elements

1. Abstract base class with common interface
2. Proper initialization with error handling
3. Health check functionality
4. Comprehensive logging throughout
5. Input validation for all methods
6. Error handling with specific exception types
7. Type hints for better code documentation
8. Async/await patterns for I/O operations
9. Private helper methods for internal logic
10. Service factory pattern for instance management
11. Consistent method signatures across CRUD operations
12. Documentation strings for all public methods

Usage Example

# Initialize and use the service
service = ExampleService()
await service.initialize()

# Create an item
item = await service.create_item({"name": "Test Item"})

# Retrieve the item
retrieved = await service.get_item(item["id"])

# Update the item
updated = await service.update_item(item["id"], {"name": "Updated Item"})

# Delete the item
success = await service.delete_item(item["id"])