Instruction file imported from alfaroukhesham/HectogonCRM (.cursor/rules/service-layer.mdc). Copyright stays with the author.

Service Layer Patterns

Service Folder Structure

Each service should be organized in its own folder with clear separation of concerns:

services/
├── service-name/
│   ├── __init__.py           # Service exports and dependencies
│   ├── service.py            # Main service class and business logic
│   ├── models.py             # Service-specific Pydantic models
│   ├── types.py              # Type definitions and enums
│   ├── utils.py              # Helper functions and utilities
│   ├── constants.py          # Service-specific constants
│   ├── test_service.py       # Unit tests for the service
│   └── integration_test.py   # Integration tests (if needed)

Service Structure

# service-name/service.py
class ServiceName:
    def __init__(self, db=None, cache_service=None):
        self.collection = db.collection_name if db else None
        self.cache_service = cache_service

    async def business_method(self, param: Type) -> ReturnType:
        # Business logic implementation
        pass

Service File Organization

init.py

# service-name/__init__.py
from fastapi import Depends
from typing import Optional
from .service import ServiceName
# Import DI providers and types from their real locations in your codebase:
# from backend.app.dependencies import get_database, get_cache_service
# from backend.app.services.cache import CacheService

async def get_service_name(
    db = Depends(get_database),
    cache_service: "CacheService" = Depends(get_cache_service),
) -> ServiceName:
    return ServiceName(repo=db, cache_service=cache_service)

__all__ = ("ServiceName", "get_service_name")

# Optional FastAPI-modern style using Annotated:
# from typing import Annotated
# async def get_service_name(
#     db: Annotated[DB, Depends(get_database)],
#     cache_service: Annotated[CacheService, Depends(get_cache_service)],
# ) -> ServiceName:
#     return ServiceName(repo=db, cache_service=cache_service)

models.py - Service-specific Models

# service-name/models.py
from pydantic import BaseModel
from typing import Optional

class ServiceSpecificModel(BaseModel):
    field1: str
    field2: Optional[int] = None

types.py - Type Definitions

# service-name/types.py
from enum import Enum

class ServiceStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

utils.py - Helper Functions

# service-name/utils.py
def format_service_data(data: dict) -> dict:
    """Helper function for data formatting"""
    return {k: v for k, v in data.items() if v is not None}

constants.py - Service Constants

# service-name/constants.py
DEFAULT_TIMEOUT = 30
MAX_RETRIES = 3
SERVICE_NAME = "service-name"

test_service.py - Unit Tests

# service-name/test_service.py
import pytest
from unittest.mock import AsyncMock
from .service import ServiceName

@pytest.mark.asyncio
async def test_business_method():
    # Arrange
    mock_db = AsyncMock()
    service = ServiceName(mock_db)

    # Act
    result = await service.business_method("test")

    # Assert
    assert result is not None

Dependency Injection

• Constructor injection: Pass database and cache services to constructors
• Optional dependencies: Make cache and other services optional for testing
• Service composition: Services can depend on other services
• Configuration: Pass settings through service constructors

Business Logic Patterns

• Separation of concerns: Keep business logic separate from route handlers
• Validation: Validate business rules in service layer
• Error handling: Handle business logic errors appropriately
• Transaction management: Use database transactions for multi-step operations

Caching Integration

# Cache-first pattern
if self.cache_service:
    cached = await self.cache_service.get_cached_data(key)
    if cached:
        return cached

# Database operation
data = await self.collection.find_one(query)

# Cache result
if self.cache_service and data:
    await self.cache_service.cache_data(key, data)

return data

Cache Invalidation

• Strategic invalidation: Invalidate cache on data changes
• Pattern-based: Use key patterns for bulk invalidation
• Organization scoping: Scope invalidation to specific organizations
• Graceful degradation: Continue operations if cache fails

Repository Pattern

• Data access abstraction: Abstract database operations
• Consistent interface: Standard CRUD operations across entities
• Query optimization: Optimize queries in service layer
• Connection management: Handle database connections properly

Error Handling

• Custom exceptions: Define domain-specific exceptions
• Logging: Log errors with appropriate context
• Recovery: Implement retry logic for transient failures
• Fallbacks: Provide fallback behavior when services are unavailable

Multi-tenant Operations

• Organization validation: Verify user access to organization
• Data filtering: Filter data by organization membership
• Permission checks: Validate permissions in service layer
• Audit logging: Track operations for compliance

Service Dependencies

async def get_service(
    db=Depends(get_database),
    cache_service: CacheService = Depends(get_cache_service)
) -> ServiceClass:
    return ServiceClass(db, cache_service)

Testing Considerations

• Mock dependencies: Make services testable with mocked dependencies
• Async testing: Use async test patterns for service methods
• Integration testing: Test service interactions with real dependencies
• Fixture management: Use fixtures for common test setup