orion/service_layer_readme.md

FastAPI Service Layer Architecture

This document describes the service layer architecture implemented in our FastAPI application, providing clear separation of concerns between HTTP handling (routers) and business logic (services).

Table of Contents

• Overview
• Architecture Pattern
• Service Layer Components
• Benefits
• Implementation Guide
• Testing Strategy
• Usage Examples
• Best Practices
• Migration Guide

Overview

Our FastAPI application follows a Service Layer Architecture pattern that separates HTTP concerns from business logic. This architecture provides better maintainability, testability, and code reusability.

Before vs After

Before (Router handling everything):

@router.post("/products")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    # Validation logic
    if product.gtin:
        normalized_gtin = normalize_gtin(product.gtin)
        if not normalized_gtin:
            raise HTTPException(status_code=400, detail="Invalid GTIN")
    
    # Business logic
    db_product = Product(**product.model_dump())
    db.add(db_product)
    db.commit()
    
    # Return response
    return db_product

After (Service layer separation):

# Router (HTTP concerns only)
@router.post("/products")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    try:
        return product_service.create_product(db, product)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

# Service (Business logic)
class ProductService:
    def create_product(self, db: Session, product_data: ProductCreate) -> Product:
        # All validation and business logic here
        # Returns domain objects, raises domain exceptions

Architecture Pattern

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   HTTP Client   │    │   FastAPI App   │    │    Database     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         │ HTTP Request           │                       │
         ├──────────────────────▶ │                       │
         │                       │                       │
         │              ┌─────────────────┐               │
         │              │    Router       │               │
         │              │  (HTTP Layer)   │               │
         │              └─────────────────┘               │
         │                       │                       │
         │                       │ Delegates to          │
         │                       ▼                       │
         │              ┌─────────────────┐               │
         │              │    Service      │               │
         │              │ (Business Layer)│               │
         │              └─────────────────┘               │
         │                       │                       │
         │                       │ Database Operations   │
         │                       ├──────────────────────▶│
         │                       │                       │
         │ HTTP Response          │                       │
         ◀──────────────────────── │                       │

Service Layer Components

1. ProductService

File: product_service.py

Responsibilities:

• Product CRUD operations
• GTIN validation and normalization
• Price processing
• Stock information retrieval
• CSV export generation

Key Methods:

class ProductService:
    def create_product(self, db: Session, product_data: ProductCreate) -> Product
    def get_product_by_id(self, db: Session, product_id: str) -> Optional[Product]
    def get_products_with_filters(self, db: Session, **filters) -> tuple[List[Product], int]
    def update_product(self, db: Session, product_id: str, product_update: ProductUpdate) -> Product
    def delete_product(self, db: Session, product_id: str) -> bool
    def get_stock_info(self, db: Session, gtin: str) -> Optional[StockSummaryResponse]
    def generate_csv_export(self, db: Session, **filters) -> Generator[str, None, None]

2. StockService

File: stock_service.py

Responsibilities:

• Stock quantity management
• GTIN normalization
• Stock location tracking
• Stock operations (set, add, remove)

Key Methods:

class StockService:
    def set_stock(self, db: Session, stock_data: StockCreate) -> Stock
    def add_stock(self, db: Session, stock_data: StockAdd) -> Stock
    def remove_stock(self, db: Session, stock_data: StockAdd) -> Stock
    def get_stock_by_gtin(self, db: Session, gtin: str) -> StockSummaryResponse
    def get_total_stock(self, db: Session, gtin: str) -> dict
    def get_all_stock(self, db: Session, **filters) -> List[Stock]
    def normalize_gtin(self, gtin_value) -> Optional[str]

3. MarketplaceService

File: marketplace_service.py

Responsibilities:

• Marketplace import job management
• Shop access validation
• User permission checking
• Job lifecycle management
• Import statistics

Key Methods:

class MarketplaceService:
    def validate_shop_access(self, db: Session, shop_code: str, user: User) -> Shop
    def create_import_job(self, db: Session, request: MarketplaceImportRequest, user: User) -> MarketplaceImportJob
    def get_import_job_by_id(self, db: Session, job_id: int, user: User) -> MarketplaceImportJob
    def get_import_jobs(self, db: Session, user: User, **filters) -> List[MarketplaceImportJob]
    def update_job_status(self, db: Session, job_id: int, status: str, **kwargs) -> MarketplaceImportJob
    def cancel_import_job(self, db: Session, job_id: int, user: User) -> MarketplaceImportJob
    def delete_import_job(self, db: Session, job_id: int, user: User) -> bool

Benefits

1. Separation of Concerns

• Routers: Handle HTTP requests/responses, authentication, validation of HTTP-specific concerns
• Services: Handle business logic, data validation, domain rules
• Models: Define data structures and database relationships

2. Better Testability

• Unit Testing: Test business logic independently of HTTP layer
• Integration Testing: Test HTTP endpoints separately from business logic
• Mocking: Easy to mock services for router testing

3. Code Reusability

• Services can be used from multiple routers
• Background tasks can use services directly
• CLI commands can leverage the same business logic

4. Maintainability

• Business logic changes only need to be made in one place
• Clear boundaries between different layers
• Easier to understand and modify code

5. Error Handling

• Domain-specific exceptions in services
• Consistent HTTP error mapping in routers
• Better error messages and logging

Implementation Guide

Step 1: Create Service Class

# product_service.py
class ProductService:
    def __init__(self):
        self.gtin_processor = GTINProcessor()
        self.price_processor = PriceProcessor()

    def create_product(self, db: Session, product_data: ProductCreate) -> Product:
        # Business logic here
        # Raise ValueError for business logic errors
        # Return domain objects
        pass

# Create singleton instance
product_service = ProductService()

Step 2: Refactor Router

# product.py
from product_service import product_service

@router.post("/products", response_model=ProductResponse)
def create_product(
    product: ProductCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    try:
        result = product_service.create_product(db, product)
        return result
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.error(f"Error creating product: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

Step 3: Error Handling Pattern

Service Layer Exceptions:

• ValueError: Business logic validation errors → HTTP 400
• PermissionError: Access control errors → HTTP 403
• Custom exceptions for specific domain errors

Router Error Translation:

try:
    result = service.method(db, data)
    return result
except ValueError as e:
    raise HTTPException(status_code=400, detail=str(e))
except PermissionError as e:
    raise HTTPException(status_code=403, detail=str(e))
except Exception as e:
    logger.error(f"Unexpected error: {str(e)}")
    raise HTTPException(status_code=500, detail="Internal server error")

Testing Strategy

Service Layer Tests

# test_product_service.py
class TestProductService:
    def setup_method(self):
        self.service = ProductService()

    def test_create_product_success(self, db):
        product_data = ProductCreate(product_id="TEST001", title="Test Product")
        result = self.service.create_product(db, product_data)
        
        assert result.product_id == "TEST001"
        assert result.title == "Test Product"

    def test_create_product_invalid_gtin(self, db):
        product_data = ProductCreate(product_id="TEST001", gtin="invalid")
        
        with pytest.raises(ValueError, match="Invalid GTIN format"):
            self.service.create_product(db, product_data)

Router Tests

# test_products_api.py
def test_create_product_endpoint(self, client, auth_headers):
    product_data = {"product_id": "TEST001", "title": "Test Product"}
    response = client.post("/api/v1/product", headers=auth_headers, json=product_data)
    
    assert response.status_code == 200
    assert response.json()["product_id"] == "TEST001"

def test_create_product_validation_error(self, client, auth_headers):
    product_data = {"product_id": "TEST001", "gtin": "invalid"}
    response = client.post("/api/v1/product", headers=auth_headers, json=product_data)
    
    assert response.status_code == 400
    assert "Invalid GTIN format" in response.json()["detail"]

Test Organization

tests/
├── test_services.py           # Service layer unit tests
├── test_product_service.py    # ProductService specific tests
├── test_stock_service.py      # StockService specific tests
├── test_marketplace_service.py # MarketplaceService specific tests
├── test_products.py           # Product API integration tests
├── test_stock.py              # Stock API integration tests
└── test_marketplace.py        # Marketplace API integration tests

Usage Examples

Using Services in Background Tasks

# background_tasks.py
from product_service import product_service
from stock_service import stock_service

async def process_marketplace_import(job_id: int, csv_data: str):
    try:
        for row in csv_data:
            # Create product using service
            product = product_service.create_product(db, product_data)
            
            # Update stock using service
            if product.gtin:
                stock_service.set_stock(db, stock_data)
                
    except Exception as e:
        marketplace_service.update_job_status(db, job_id, "failed", error_message=str(e))

Using Services in CLI Commands

# cli_commands.py
from product_service import product_service

def bulk_update_prices(csv_file: str):
    """CLI command to bulk update product prices"""
    for product_id, new_price in read_csv(csv_file):
        try:
            product_service.update_product(
                db, product_id, 
                ProductUpdate(price=new_price)
            )
            print(f"Updated {product_id}")
        except ValueError as e:
            print(f"Error updating {product_id}: {e}")

Service Composition

# order_service.py (hypothetical)
from product_service import product_service
from stock_service import stock_service

class OrderService:
    def create_order(self, db: Session, order_data: OrderCreate) -> Order:
        for item in order_data.items:
            # Check product exists
            product = product_service.get_product_by_id(db, item.product_id)
            if not product:
                raise ValueError(f"Product {item.product_id} not found")
            
            # Check stock availability
            stock = stock_service.get_total_stock(db, product.gtin)
            if stock["total_quantity"] < item.quantity:
                raise ValueError(f"Insufficient stock for {item.product_id}")
            
            # Reserve stock
            stock_service.remove_stock(db, StockAdd(
                gtin=product.gtin,
                location="RESERVED",
                quantity=item.quantity
            ))

Best Practices

1. Service Design Principles

• Single Responsibility: Each service handles one domain
• Dependency Injection: Services should not create their own dependencies
• Pure Functions: Methods should be predictable and side-effect free when possible
• Domain Exceptions: Use meaningful exception types

2. Error Handling

# Good: Specific domain exceptions
class ProductService:
    def create_product(self, db: Session, product_data: ProductCreate) -> Product:
        if self.product_exists(db, product_data.product_id):
            raise ValueError(f"Product {product_data.product_id} already exists")
        
        if product_data.gtin and not self.is_valid_gtin(product_data.gtin):
            raise ValueError("Invalid GTIN format")
        
        # Business logic...

# Bad: Generic exceptions
def create_product(product_data):
    if product_exists(product_data.product_id):
        raise Exception("Error occurred")  # Too generic

3. Testing Guidelines

• Test business logic in service tests
• Test HTTP behavior in router tests
• Use fixtures for common test data
• Mock external dependencies

4. Documentation

• Document service methods with docstrings
• Include parameter types and return types
• Document exceptions that can be raised
• Provide usage examples

def create_product(self, db: Session, product_data: ProductCreate) -> Product:
    """Create a new product with validation.
    
    Args:
        db: Database session
        product_data: Product creation data
        
    Returns:
        Created Product instance
        
    Raises:
        ValueError: If product_id already exists or GTIN is invalid
        
    Example:
        >>> service = ProductService()
        >>> product = service.create_product(db, ProductCreate(
        ...     product_id="ABC123",
        ...     title="Test Product",
        ...     gtin="1234567890123"
        ... ))
    """

Migration Guide

Migrating Existing Routers

1. Identify Business Logic: Look for code that validates data, processes business rules, or manipulates domain objects

2. Create Service Class: Extract business logic into service methods

3. Update Router: Replace business logic with service calls and add error handling

4. Add Tests: Create comprehensive tests for the service layer

5. Verify Compatibility: Ensure API behavior remains the same

Example Migration

Before:

@router.post("/products")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    # Check if product exists
    existing = db.query(Product).filter(Product.product_id == product.product_id).first()
    if existing:
        raise HTTPException(status_code=400, detail="Product already exists")
    
    # Validate GTIN
    if product.gtin:
        normalized_gtin = normalize_gtin(product.gtin)
        if not normalized_gtin:
            raise HTTPException(status_code=400, detail="Invalid GTIN")
        product.gtin = normalized_gtin
    
    # Create product
    db_product = Product(**product.model_dump())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    
    return db_product

After:

# Service
class ProductService:
    def create_product(self, db: Session, product_data: ProductCreate) -> Product:
        # Check if product exists
        if self.product_exists(db, product_data.product_id):
            raise ValueError("Product already exists")
        
        # Validate GTIN
        if product_data.gtin:
            normalized_gtin = self.gtin_processor.normalize(product_data.gtin)
            if not normalized_gtin:
                raise ValueError("Invalid GTIN")
            product_data.gtin = normalized_gtin
        
        # Create product
        db_product = Product(**product_data.model_dump())
        db.add(db_product)
        db.commit()
        db.refresh(db_product)
        
        return db_product

# Router
@router.post("/products")
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    try:
        return product_service.create_product(db, product)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.error(f"Error creating product: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

Conclusion

The service layer architecture provides a robust foundation for scalable FastAPI applications. By separating HTTP concerns from business logic, we achieve better maintainability, testability, and code reuse. This pattern scales well as applications grow and makes it easier to add new features while maintaining existing functionality.

For questions or contributions to this architecture, please refer to the team documentation or create an issue in the project repository.