Services

Services are the business logic layer of Tasteful flavors. They contain all the core business rules, orchestrate operations between different components, and provide a clean interface for controllers to interact with your application’s functionality. Services are where ALL business logic should live.

What is a Service?

A Service in Tasteful is a class that inherits from BaseService and encapsulates business logic. Services are responsible for:

Business Logic Implementation: All core business rules and operations
Data Orchestration: Coordinating operations between repositories and external systems
Validation: Ensuring data integrity and business rule compliance
Transaction Management: Handling complex operations that span multiple data sources
Error Handling: Converting low-level errors into meaningful business exceptions
Service Composition: Coordinating with other services for complex workflows

Services act as the single source of truth for business operations and should contain all logic that isn’t purely HTTP-related (controllers) or data-access related (repositories).

Core Concepts

BaseService Foundation

All services inherit from BaseService, which provides the foundation for dependency injection:

from tasteful.base_flavor import BaseService

class UserService(BaseService):
    def __init__(self, user_repository: UserRepository):
        super().__init__()
        self.user_repository = user_repository

Key features of BaseService:

Dependency Injection Support: Automatically receives dependencies through constructor injection
Clean Architecture: Provides a clear separation between business logic and infrastructure
Testability: Easy to mock and test in isolation
Composition: Services can depend on other services and repositories

Service Dependencies

Services can depend on repositories, other services, and configuration objects:

class OrderService(BaseService):
    def __init__(
        self,
        order_repository: OrderRepository,
        user_service: UserService,
        payment_service: PaymentService,
        order_config: OrderConfig
    ):
        super().__init__()
        self.order_repository = order_repository
        self.user_service = user_service
        self.payment_service = payment_service
        self.order_config = order_config

Implementation

Basic Service Implementation

Here’s a complete example of a service implementation:

from tasteful.base_flavor import BaseService
from typing import Dict, List, Optional

class ProductService(BaseService):
    """Service for product management business logic."""
    
    def __init__(self, product_repository: ProductRepository):
        super().__init__()
        self.product_repository = product_repository
    
    def list_products(self, skip: int = 0, limit: int = 100) -> List[Dict]:
        """List products with business logic applied."""
        # Business logic: Apply default sorting and filtering
        products = self.product_repository.list_products(
            skip=skip, 
            limit=min(limit, 1000),  # Business rule: max 1000 items
            order_by="created_at",
            active_only=True
        )
        
        # Business logic: Add computed fields
        for product in products:
            product["display_price"] = self._format_price(product["price"])
            product["in_stock"] = product["quantity"] > 0
        
        return products
    
    def get_product(self, product_id: int) -> Optional[Dict]:
        """Get a product with business logic validation."""
        if product_id <= 0:
            raise ValueError("Product ID must be positive")
        
        product = self.product_repository.get_by_id(product_id)
        if not product:
            raise ProductNotFoundError(f"Product {product_id} not found")
        
        # Business logic: Add computed fields
        product["display_price"] = self._format_price(product["price"])
        product["availability_status"] = self._get_availability_status(product)
        
        return product
    
    def create_product(self, product_data: Dict) -> Dict:
        """Create a new product with validation."""
        # Business validation
        self._validate_product_data(product_data)
        
        # Business logic: Set defaults
        product_data["status"] = "active"
        product_data["created_at"] = datetime.utcnow()
        
        # Business logic: Generate SKU if not provided
        if not product_data.get("sku"):
            product_data["sku"] = self._generate_sku(product_data["name"])
        
        return self.product_repository.create(product_data)
    
    def update_product(self, product_id: int, product_data: Dict) -> Dict:
        """Update a product with business validation."""
        # Check if product exists
        existing_product = self.get_product(product_id)
        
        # Business validation
        self._validate_product_update(existing_product, product_data)
        
        # Business logic: Update timestamp
        product_data["updated_at"] = datetime.utcnow()
        
        return self.product_repository.update(product_id, product_data)
    
    def delete_product(self, product_id: int) -> Dict:
        """Soft delete a product with business rules."""
        product = self.get_product(product_id)
        
        # Business rule: Can't delete products with active orders
        if self._has_active_orders(product_id):
            raise BusinessRuleError("Cannot delete product with active orders")
        
        # Business logic: Soft delete instead of hard delete
        return self.product_repository.update(product_id, {
            "status": "deleted",
            "deleted_at": datetime.utcnow()
        })
    
    def _validate_product_data(self, product_data: Dict) -> None:
        """Private method for product validation."""
        if not product_data.get("name"):
            raise ValidationError("Product name is required")
        
        if product_data.get("price", 0) < 0:
            raise ValidationError("Product price cannot be negative")
    
    def _format_price(self, price: float) -> str:
        """Format price for display."""
        return f"${price:.2f}"
    
    def _get_availability_status(self, product: Dict) -> str:
        """Determine product availability status."""
        if product["quantity"] == 0:
            return "out_of_stock"
        elif product["quantity"] < 10:
            return "low_stock"
        else:
            return "in_stock"

Advanced Patterns

Service Composition

Services can orchestrate complex operations by coordinating with multiple other services:

class OrderService(BaseService):
    def __init__(
        self,
        order_repository: OrderRepository,
        user_service: UserService,
        product_service: ProductService,
        payment_service: PaymentService,
        inventory_service: InventoryService,
        notification_service: NotificationService
    ):
        super().__init__()
        self.order_repository = order_repository
        self.user_service = user_service
        self.product_service = product_service
        self.payment_service = payment_service
        self.inventory_service = inventory_service
        self.notification_service = notification_service
    
    def create_order(self, order_data: Dict) -> Dict:
        """Create an order with complex business logic."""
        # Step 1: Validate user
        user = self.user_service.get_user(order_data["user_id"])
        if not user["is_active"]:
            raise BusinessRuleError("User account is not active")
        
        # Step 2: Validate products and calculate totals
        total_amount = 0
        for item in order_data["items"]:
            product = self.product_service.get_product(item["product_id"])
            if not product:
                raise ValidationError(f"Product {item['product_id']} not found")
            
            # Business logic: Apply discounts, calculate totals
            item_total = self._calculate_item_total(product, item["quantity"])
            total_amount += item_total
        
        # Step 3: Check inventory availability
        inventory_check = self.inventory_service.check_availability(order_data["items"])
        if not inventory_check["available"]:
            raise BusinessRuleError("Some items are not available")
        
        # Step 4: Create order record
        order_data.update({
            "total_amount": total_amount,
            "status": "pending",
            "created_at": datetime.utcnow()
        })
        order = self.order_repository.create(order_data)
        
        # Step 5: Reserve inventory
        self.inventory_service.reserve_items(order["id"], order_data["items"])
        
        # Step 6: Send confirmation
        self.notification_service.send_order_confirmation(user["email"], order)
        
        return order
    
    def process_payment(self, order_id: int, payment_data: Dict) -> Dict:
        """Process payment for an order."""
        order = self.order_repository.get_by_id(order_id)
        if not order:
            raise OrderNotFoundError(f"Order {order_id} not found")
        
        if order["status"] != "pending":
            raise BusinessRuleError("Order is not in pending status")
        
        try:
            # Process payment through payment service
            payment_result = self.payment_service.process_payment(
                amount=order["total_amount"],
                payment_data=payment_data
            )
            
            if payment_result["success"]:
                # Update order status
                self.order_repository.update(order_id, {
                    "status": "paid",
                    "payment_id": payment_result["payment_id"],
                    "paid_at": datetime.utcnow()
                })
                
                # Confirm inventory reservation
                self.inventory_service.confirm_reservation(order_id)
                
                # Send success notification
                user = self.user_service.get_user(order["user_id"])
                self.notification_service.send_payment_success(user["email"], order)
                
                return {"success": True, "order": order}
            else:
                # Release inventory reservation
                self.inventory_service.release_reservation(order_id)
                return {"success": False, "error": payment_result["error"]}
                
        except Exception as e:
            # Release inventory on any error
            self.inventory_service.release_reservation(order_id)
            raise PaymentProcessingError(f"Payment processing failed: {str(e)}")

Service with Repository Integration

Services should always interact with data through repositories:

class UserService(BaseService):
    def __init__(self, user_repository: UserRepository, auth_service: AuthService):
        super().__init__()
        self.user_repository = user_repository
        self.auth_service = auth_service
    
    def create_user(self, user_data: Dict) -> Dict:
        """Create a new user with business validation."""
        # Business validation
        self._validate_user_data(user_data)
        
        # Business rule: Check if email already exists
        if self.user_repository.email_exists(user_data["email"]):
            raise BusinessRuleError("Email already exists")
        
        # Business logic: Hash password
        if "password" in user_data:
            user_data["password_hash"] = self.auth_service.hash_password(user_data["password"])
            del user_data["password"]  # Remove plain password
        
        # Business logic: Set defaults
        user_data.update({
            "is_active": True,
            "created_at": datetime.utcnow(),
            "email_verified": False
        })
        
        # Create user through repository
        user = self.user_repository.create(user_data)
        
        # Business logic: Send welcome email
        self._send_welcome_email(user)
        
        return user
    
    def authenticate_user(self, email: str, password: str) -> Optional[Dict]:
        """Authenticate a user with business logic."""
        user = self.user_repository.get_by_email(email)
        if not user:
            return None
        
        # Business rule: Check if account is active
        if not user["is_active"]:
            raise BusinessRuleError("Account is deactivated")
        
        # Verify password
        if not self.auth_service.verify_password(password, user["password_hash"]):
            # Business logic: Track failed login attempts
            self._track_failed_login(user["id"])
            return None
        
        # Business logic: Update last login
        self.user_repository.update(user["id"], {
            "last_login": datetime.utcnow(),
            "failed_login_attempts": 0
        })
        
        return user

Integration with Other Components

Service-Controller Relationship

Controllers should always delegate business logic to services:

# In Controller
class UserController(BaseController):
    def __init__(self, user_service: UserService):
        super().__init__(prefix="/users", tags=["users"])
        self.user_service = user_service
    
    @Post("/")
    async def create_user(self, user_data: Dict) -> Dict:
        # Controller delegates to service
        return self.user_service.create_user(user_data)
    
    @Post("/login")
    async def login(self, credentials: Dict) -> Dict:
        # Controller handles HTTP, service handles business logic
        user = self.user_service.authenticate_user(
            credentials["email"], 
            credentials["password"]
        )
        if user:
            return {"success": True, "user": user}
        else:
            raise HTTPException(status_code=401, detail="Invalid credentials")

Service-Repository Relationship

Services should interact with data only through repositories:

# ✅ Good: Service uses repository for data access
class ProductService(BaseService):
    def __init__(self, product_repository: ProductRepository):
        super().__init__()
        self.product_repository = product_repository
    
    def get_product(self, product_id: int) -> Dict:
        # Service uses repository for data access
        product = self.product_repository.get_by_id(product_id)
        if not product:
            raise ProductNotFoundError("Product not found")
        
        # Service adds business logic
        product["display_price"] = self._format_price(product["price"])
        return product

# ❌ Bad: Service directly accessing database
class ProductService(BaseService):
    def __init__(self, database_connection):
        super().__init__()
        self.db = database_connection
    
    def get_product(self, product_id: int) -> Dict:
        # Don't access database directly from service
        result = self.db.execute("SELECT * FROM products WHERE id = ?", product_id)
        return result.fetchone()

Dependency Injection

Services receive their dependencies through constructor injection:

class OrderService(BaseService):
    def __init__(
        self,
        order_repository: OrderRepository,    # ← Repository dependency
        user_service: UserService,           # ← Service dependency
        payment_service: PaymentService,     # ← Service dependency
        order_config: OrderConfig            # ← Configuration dependency
    ):
        super().__init__()
        self.order_repository = order_repository
        self.user_service = user_service
        self.payment_service = payment_service
        self.order_config = order_config

# In your flavor definition
class OrderFlavor(BaseFlavor):
    def __init__(self):
        super().__init__(
            controller=OrderController,
            services=[
                OrderService,
                UserService,
                PaymentService
            ],  # ← Services registered for injection
            repositories=[OrderRepository, UserRepository],
            config=OrderConfig
        )

Best Practices

Do’s

Put ALL business logic in services - Never put business logic in controllers or repositories
Use meaningful method names - Service methods should clearly describe what business operation they perform
Validate input data - Services should validate all input according to business rules
Handle errors appropriately - Convert low-level errors into meaningful business exceptions
Keep services focused - Each service should have a single responsibility
Use dependency injection - Let the framework inject dependencies rather than creating them manually

class UserService(BaseService):
    def __init__(self, user_repository: UserRepository, email_service: EmailService):
        super().__init__()
        self.user_repository = user_repository
        self.email_service = email_service
    
    def create_user(self, user_data: Dict) -> Dict:
        """Create a new user with full business logic."""
        # Validate business rules
        self._validate_user_creation(user_data)
        
        # Apply business logic
        user_data["status"] = "active"
        user_data["created_at"] = datetime.utcnow()
        
        # Create through repository
        user = self.user_repository.create(user_data)
        
        # Handle side effects
        self.email_service.send_welcome_email(user["email"])
        
        return user
    
    def _validate_user_creation(self, user_data: Dict) -> None:
        """Private validation method."""
        if not user_data.get("email"):
            raise ValidationError("Email is required")
        
        if self.user_repository.email_exists(user_data["email"]):
            raise BusinessRuleError("Email already exists")

Don’ts

Don’t put HTTP logic in services - Services shouldn’t know about HTTP requests/responses
Don’t access external APIs directly - Use dedicated service classes for external integrations
Don’t handle database connections directly - Always use repositories for data access
Don’t make services too large - Break down complex services into smaller, focused services
Don’t ignore error handling - Always handle and convert exceptions appropriately

# ❌ Bad: HTTP logic in service
class UserService(BaseService):
    def create_user(self, request: HTTPRequest) -> HTTPResponse:
        # Don't handle HTTP in services
        user_data = request.json()
        user = self.user_repository.create(user_data)
        return HTTPResponse(json=user, status=201)

# ❌ Bad: Direct database access
class UserService(BaseService):
    def __init__(self, database_url: str):
        self.db = create_engine(database_url)
    
    def get_user(self, user_id: int):
        # Don't access database directly
        with self.db.connect() as conn:
            result = conn.execute("SELECT * FROM users WHERE id = ?", user_id)
            return result.fetchone()

# ✅ Good: Clean service with proper separation
class UserService(BaseService):
    def __init__(self, user_repository: UserRepository):
        super().__init__()
        self.user_repository = user_repository
    
    def create_user(self, user_data: Dict) -> Dict:
        # Pure business logic
        self._validate_user_data(user_data)
        return self.user_repository.create(user_data)

Common Patterns

CRUD Service Pattern

Most services follow standard CRUD patterns with business logic:

class ResourceService(BaseService):
    def __init__(self, resource_repository: ResourceRepository):
        super().__init__()
        self.resource_repository = resource_repository
    
    def list_resources(self, filters: Dict = None, pagination: Dict = None) -> List[Dict]:
        """List resources with business filtering."""
        # Apply business rules to filters
        safe_filters = self._sanitize_filters(filters or {})
        safe_pagination = self._validate_pagination(pagination or {})
        
        return self.resource_repository.list(safe_filters, safe_pagination)
    
    def get_resource(self, resource_id: int) -> Dict:
        """Get a resource with business validation."""
        if resource_id <= 0:
            raise ValidationError("Invalid resource ID")
        
        resource = self.resource_repository.get_by_id(resource_id)
        if not resource:
            raise ResourceNotFoundError("Resource not found")
        
        return resource
    
    def create_resource(self, resource_data: Dict) -> Dict:
        """Create a resource with business validation."""
        self._validate_resource_data(resource_data)
        
        # Apply business defaults
        resource_data["status"] = "active"
        resource_data["created_at"] = datetime.utcnow()
        
        return self.resource_repository.create(resource_data)
    
    def update_resource(self, resource_id: int, resource_data: Dict) -> Dict:
        """Update a resource with business validation."""
        existing_resource = self.get_resource(resource_id)
        
        # Business validation for updates
        self._validate_resource_update(existing_resource, resource_data)
        
        resource_data["updated_at"] = datetime.utcnow()
        return self.resource_repository.update(resource_id, resource_data)
    
    def delete_resource(self, resource_id: int) -> bool:
        """Delete a resource with business rules."""
        resource = self.get_resource(resource_id)
        
        # Business rule: Check if resource can be deleted
        if self._has_dependencies(resource_id):
            raise BusinessRuleError("Cannot delete resource with dependencies")
        
        return self.resource_repository.delete(resource_id)

Service Orchestration Pattern

Services can orchestrate complex workflows:

class CheckoutService(BaseService):
    def __init__(
        self,
        cart_service: CartService,
        inventory_service: InventoryService,
        payment_service: PaymentService,
        order_service: OrderService,
        notification_service: NotificationService
    ):
        super().__init__()
        self.cart_service = cart_service
        self.inventory_service = inventory_service
        self.payment_service = payment_service
        self.order_service = order_service
        self.notification_service = notification_service
    
    def process_checkout(self, user_id: int, payment_data: Dict) -> Dict:
        """Process a complete checkout workflow."""
        try:
            # Step 1: Get cart
            cart = self.cart_service.get_user_cart(user_id)
            if not cart["items"]:
                raise BusinessRuleError("Cart is empty")
            
            # Step 2: Validate inventory
            inventory_check = self.inventory_service.check_availability(cart["items"])
            if not inventory_check["available"]:
                raise BusinessRuleError("Some items are no longer available")
            
            # Step 3: Create order
            order = self.order_service.create_order_from_cart(cart)
            
            # Step 4: Process payment
            payment_result = self.payment_service.process_payment(
                order["total"], payment_data
            )
            
            if not payment_result["success"]:
                # Rollback order
                self.order_service.cancel_order(order["id"])
                raise PaymentError("Payment failed")
            
            # Step 5: Confirm order
            self.order_service.confirm_order(order["id"], payment_result["payment_id"])
            
            # Step 6: Update inventory
            self.inventory_service.reserve_items(order["items"])
            
            # Step 7: Clear cart
            self.cart_service.clear_cart(user_id)
            
            # Step 8: Send notifications
            self.notification_service.send_order_confirmation(user_id, order)
            
            return {
                "success": True,
                "order_id": order["id"],
                "payment_id": payment_result["payment_id"]
            }
            
        except Exception as e:
            # Handle any errors and rollback if necessary
            self._handle_checkout_error(e, locals())
            raise

Testing

Services are highly testable thanks to dependency injection:

import pytest
from unittest.mock import Mock, patch
from datetime import datetime

def test_user_service_create_user():
    # Arrange
    mock_user_repository = Mock(spec=UserRepository)
    mock_user_repository.email_exists.return_value = False
    mock_user_repository.create.return_value = {"id": 1, "email": "test@example.com"}
    
    mock_email_service = Mock(spec=EmailService)
    
    user_service = UserService(
        user_repository=mock_user_repository,
        email_service=mock_email_service
    )
    
    user_data = {"email": "test@example.com", "name": "Test User"}
    
    # Act
    result = user_service.create_user(user_data)
    
    # Assert
    assert result["id"] == 1
    assert result["email"] == "test@example.com"
    mock_user_repository.email_exists.assert_called_once_with("test@example.com")
    mock_user_repository.create.assert_called_once()
    mock_email_service.send_welcome_email.assert_called_once_with("test@example.com")

def test_user_service_create_user_duplicate_email():
    # Arrange
    mock_user_repository = Mock(spec=UserRepository)
    mock_user_repository.email_exists.return_value = True
    
    user_service = UserService(user_repository=mock_user_repository)
    user_data = {"email": "test@example.com", "name": "Test User"}
    
    # Act & Assert
    with pytest.raises(BusinessRuleError, match="Email already exists"):
        user_service.create_user(user_data)
    
    mock_user_repository.create.assert_not_called()

def test_order_service_complex_workflow():
    # Arrange
    mock_order_repository = Mock(spec=OrderRepository)
    mock_user_service = Mock(spec=UserService)
    mock_payment_service = Mock(spec=PaymentService)
    
    mock_user_service.get_user.return_value = {"id": 1, "is_active": True}
    mock_payment_service.process_payment.return_value = {"success": True, "payment_id": "pay_123"}
    mock_order_repository.create.return_value = {"id": 1, "total_amount": 100.0}
    
    order_service = OrderService(
        order_repository=mock_order_repository,
        user_service=mock_user_service,
        payment_service=mock_payment_service
    )
    
    order_data = {"user_id": 1, "items": [{"product_id": 1, "quantity": 2}]}
    
    # Act
    result = order_service.create_order(order_data)
    
    # Assert
    assert result["id"] == 1
    mock_user_service.get_user.assert_called_once_with(1)
    mock_order_repository.create.assert_called_once()

Services are the heart of your Tasteful application’s business logic. They provide a clean, testable, and maintainable way to implement complex business operations while maintaining proper separation of concerns with controllers and repositories.

Core Concepts

Flavors Concepts

Documentation Index

​Services

​What is a Service?

​Core Concepts

​BaseService Foundation

​Service Dependencies

​Implementation

​Basic Service Implementation

​Advanced Patterns

​Service Composition

​Service with Repository Integration

​Integration with Other Components

​Service-Controller Relationship

​Service-Repository Relationship

​Dependency Injection

​Best Practices

​Do’s

​Don’ts

​Common Patterns

​CRUD Service Pattern

​Service Orchestration Pattern

​Testing

Services

What is a Service?

Core Concepts

BaseService Foundation

Service Dependencies

Implementation

Basic Service Implementation

Advanced Patterns

Service Composition

Service with Repository Integration

Integration with Other Components

Service-Controller Relationship

Service-Repository Relationship

Dependency Injection

Best Practices

Do’s

Don’ts

Common Patterns

CRUD Service Pattern

Service Orchestration Pattern

Testing