Building Service Layers for DynamoDB Applications with boto3-assist

Published October 18, 2025 · 10 min read · by Eric Wilson · in python

Learn how to create clean, testable service layers that encapsulate business logic and database operations. Complete guide to CRUD operations, query patterns, and best practices.

In my previous posts, we've covered single-table design, one-to-many relationships, and defining models. Now let's bring it all together by building clean, maintainable service layers.

The service layer is where your application's business logic lives. It's the bridge between your API handlers (Lambda functions, REST endpoints, etc.) and your DynamoDB database. Services use models to interact with data while keeping all database operations isolated and testable.

Why Service Layers?

Separation of Concerns

API Handler → Service Layer → DynamoDB
     ↓              ↓
  Handles      Handles
  HTTP       Business Logic
              & Database

Benefits:

✅ Testability: Easy to mock database calls
✅ Reusability: Same service used by multiple handlers
✅ Maintainability: Business logic in one place
✅ Type Safety: Clear interfaces and contracts

Without Service Layer (❌ Don't Do This)

# Lambda handler with embedded database logic
def lambda_handler(event, context):
    db = DynamoDB()
    product = Product(id="123", name="Widget")
    item = product.to_resource_dictionary()
    db.save(item=item, table_name="products")
    # Business logic mixed with infrastructure

With Service Layer (✅ Do This)

# Lambda handler
def lambda_handler(event, context):
    service = ProductService()
    product = service.create_product({"id": "123", "name": "Widget"})
    return {"statusCode": 200, "body": product.to_dictionary()}

# Service handles the database
class ProductService:
    def create_product(self, data: dict) -> Product:
        product = Product().map(data)
        # Business logic and database interaction here
        item = product.to_resource_dictionary()
        self.db.save(item=item, table_name=self.table_name)
        return product

Clean separation makes your code maintainable and testable!

Basic Service Structure

Every service follows this pattern:

import os
from typing import Optional
from boto3_assist.dynamodb.dynamodb import DynamoDB
from your_app.models.product_model import Product

class ProductService:
    def __init__(self, db: Optional[DynamoDB] = None):
        """
        Initialize the service.
        
        Args:
            db: Optional DynamoDB instance (for dependency injection/testing)
        """
        self.db = db or DynamoDB()
        self.table_name = os.environ.get("APP_TABLE_NAME", "app-table")

Key Elements:

DynamoDB instance: Injected or created (enables testing)
Table name: From environment variable (12-factor app)
Dependency injection: Accepts db parameter for mocking

CRUD Operations

Create (Save)

def create_product(self, product_data: dict) -> Product:
    """
    Create a new product.
    
    Args:
        product_data: Dictionary with product attributes
        
    Returns:
        Product: The created product model
    """
    # Map data to model
    product = Product().map(product_data)
    
    # Add business logic here
    if not product.name:
        raise ValueError("Product name is required")
    
    # Convert to DynamoDB format (includes all keys)
    item = product.to_resource_dictionary()
    
    # Save to database
    self.db.save(item=item, table_name=self.table_name)
    
    return product

Usage:

service = ProductService()
product = service.create_product({
    "id": "prod-123",
    "name": "Widget",
    "price": 29.99
})

Read (Get by ID)

def get_product(self, product_id: str) -> Optional[Product]:
    """
    Retrieve a product by ID.
    
    Args:
        product_id: The product ID
        
    Returns:
        Product if found, None otherwise
    """
    # Create model with ID to identify the key
    model = Product(id=product_id)
    
    # Get from database using model's primary key
    response = self.db.get(model=model, table_name=self.table_name, do_projections=False)
    
    # Check if item exists
    item = response.get("Item")
    if not item:
        return None
    
    # Map response to model and return
    return Product().map(item)

Usage:

product = service.get_product("prod-123")
if product:
    print(f"Found: {product.name}")
else:
    print("Product not found")

Update

def update_product(self, product_id: str, updates: dict) -> Optional[Product]:
    """
    Update an existing product.
    
    Args:
        product_id: The product ID
        updates: Dictionary of fields to update
        
    Returns:
        Updated product if found, None otherwise
    """
    # 1. Get existing product
    existing = self.get_product(product_id)
    if not existing:
        return None
    
    # 2. Apply updates to the model
    existing.map(updates)
    
    # 3. Add business logic/validation
    if existing.price < 0:
        raise ValueError("Price cannot be negative")
    
    # 4. Save back to database
    item = existing.to_resource_dictionary()
    self.db.save(item=item, table_name=self.table_name)
    
    return existing

Usage:

updated = service.update_product("prod-123", {"price": 34.99})
if updated:
    print(f"Updated price: ${updated.price}")

Delete

def delete_product(self, product_id: str) -> bool:
    """
    Delete a product.
    
    Args:
        product_id: The product ID
        
    Returns:
        True if deleted successfully, False otherwise
    """
    model = Product(id=product_id)
    
    try:
        self.db.delete(model=model, table_name=self.table_name)
        return True
    except Exception as e:
        # Log the error (use proper logging in production)
        print(f"Error deleting product {product_id}: {e}")
        return False

List Operations (Queries)

List All Items (Using GSI)

from boto3.dynamodb.conditions import Key

def list_all_products(self, ascending: bool = True) -> list[Product]:
    """
    List all products, sorted by name.
    
    Args:
        ascending: Sort order
        
    Returns:
        List of Product models
    """
    # Create a model to get the GSI key
    model = Product()
    
    # Get the key condition for GSI0 (all products)
    key_condition = model.indexes.get("gsi0").key()
    
    # Query the GSI
    response = self.db.query(
        key=key_condition,
        index_name="gsi0",
        table_name=self.table_name,
        ascending=ascending
    )
    
    # Map all items to models
    items = response.get("Items", [])
    return [Product().map(item) for item in items]

Usage:

products = service.list_all_products()
for product in products:
    print(f"{product.name}: ${product.price}")

List with Filter

def list_products_by_category(self, category: str) -> list[Product]:
    """
    List products in a specific category.
    
    Assumes Product has a gsi1 with:
    - gsi1_pk = category#<category>
    - gsi1_sk = product#<id>
    """
    model = Product()
    model.category = category
    
    # Get key condition for this category
    key_condition = model.indexes.get("gsi1").key()
    
    response = self.db.query(
        key=key_condition,
        index_name="gsi1",
        table_name=self.table_name
    )
    
    items = response.get("Items", [])
    return [Product().map(item) for item in items]

Advanced: One-to-Many Relationships

Remember our Order → OrderItem example? Here's how to implement it in a service:

def get_order_with_items(self, order_id: str) -> dict:
    """
    Get an order and all its items in a single query.
    
    Returns:
        Dictionary with 'order' and 'items' keys
    """
    model = Order(id=order_id)
    
    # Query by partition key only (gets order + all items)
    key_condition = model.indexes.primary.key(include_sort_key=False)
    
    response = self.db.query(
        key=key_condition,
        table_name=self.table_name
    )
    
    items = response.get("Items", [])
    
    # Separate order from order items
    order = None
    order_items = []
    
    for item in items:
        # Check the sort key to determine type
        if item.get("sk", "").startswith("order#"):
            order = Order().map(item)
        elif item.get("sk", "").startswith("item#"):
            order_items.append(OrderItem().map(item))
    
    return {
        "order": order,
        "items": order_items
    }

Usage:

result = service.get_order_with_items("order-123")
print(f"Order total: ${result['order'].total}")
print(f"Item count: {len(result['items'])}")

This retrieves the order and ALL its items in one query—the power of single-table design!

Business Logic in Services

Services are perfect for business rules:

class OrderService:
    def __init__(self, db=None):
        self.db = db or DynamoDB()
        self.table_name = os.environ.get("APP_TABLE_NAME", "app-table")
        self.product_service = ProductService(db=self.db)
    
    def create_order(self, user_id: str, items: list[dict]) -> Order:
        """
        Create an order with business logic.
        """
        # Validate inputs
        if not items:
            raise ValueError("Order must have at least one item")
        
        # Create order
        order = Order()
        order.id = self._generate_order_id()
        order.user_id = user_id
        order.created_utc = datetime.utcnow()
        order.status = "pending"
        
        # Calculate totals
        subtotal = 0.0
        tax_total = 0.0
        
        for item_data in items:
            # Get product to check price and availability
            product = self.product_service.get_product(item_data["product_id"])
            if not product:
                raise ValueError(f"Product {item_data['product_id']} not found")
            
            quantity = item_data["quantity"]
            item_total = product.price * quantity
            subtotal += item_total
            
            if product.is_taxable:
                tax_total += item_total * 0.08  # 8% tax
        
        order.subtotal = subtotal
        order.tax_total = tax_total
        order.total = subtotal + tax_total
        
        # Save order
        self.db.save(
            item=order.to_resource_dictionary(),
            table_name=self.table_name
        )
        
        # Create order items
        for item_data in items:
            self._create_order_item(order.id, item_data)
        
        return order
    
    def _generate_order_id(self) -> str:
        """Generate unique order ID"""
        import uuid
        return f"ord-{uuid.uuid4().hex[:12]}"
    
    def _create_order_item(self, order_id: str, item_data: dict):
        """Create an order item"""
        item = OrderItem()
        item.id = f"item-{uuid.uuid4().hex[:8]}"
        item.order_id = order_id
        item.product_id = item_data["product_id"]
        item.quantity = item_data["quantity"]
        
        self.db.save(
            item=item.to_resource_dictionary(),
            table_name=self.table_name
        )

Error Handling

Always handle DynamoDB errors gracefully:

from botocore.exceptions import ClientError

def get_product(self, product_id: str) -> Optional[Product]:
    """Get product with error handling."""
    try:
        model = Product(id=product_id)
        response = self.db.get(model=model, table_name=self.table_name, do_projections=False)
        
        item = response.get("Item")
        return Product().map(item) if item else None
        
    except ClientError as e:
        error_code = e.response['Error']['Code']
        
        if error_code == 'ResourceNotFoundException':
            print(f"Table {self.table_name} not found")
        elif error_code == 'ProvisionedThroughputExceededException':
            print("Request rate too high")
        else:
            print(f"DynamoDB error: {error_code}")
        
        return None
    
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

Complete Service Example

Here's a production-ready service with all CRUD operations:

import os
from typing import Optional
from boto3_assist.dynamodb.dynamodb import DynamoDB
from boto3.dynamodb.conditions import Key
from your_app.models.user_model import User

class UserService:
    def __init__(self, db: Optional[DynamoDB] = None):
        self.db = db or DynamoDB()
        self.table_name = os.environ.get("APP_TABLE_NAME", "app-table")
    
    def create_user(self, user_data: dict) -> User:
        """Create a new user."""
        user = User().map(user_data)
        
        # Set default values
        if not user.status:
            user.status = "active"
        
        # Business logic: validate email
        if not self._is_valid_email(user.email):
            raise ValueError("Invalid email address")
        
        # Check if email already exists
        if self.get_user_by_email(user.email):
            raise ValueError("Email already registered")
        
        # Save user
        item = user.to_resource_dictionary()
        self.db.save(item=item, table_name=self.table_name)
        
        return user
    
    def get_user(self, user_id: str) -> Optional[User]:
        """Get user by ID."""
        model = User(id=user_id)
        response = self.db.get(model=model, table_name=self.table_name, do_projections=False)
        
        item = response.get("Item")
        return User().map(item) if item else None
    
    def get_user_by_email(self, email: str) -> Optional[User]:
        """
        Find user by email.
        Assumes gsi2 with: gsi2_pk = emails, gsi2_sk = email#<email>
        """
        model = User()
        model.email = email
        
        key_condition = model.indexes.get("gsi2").key()
        
        response = self.db.query(
            key=key_condition,
            index_name="gsi2",
            table_name=self.table_name
        )
        
        items = response.get("Items", [])
        return User().map(items[0]) if items else None
    
    def list_users(self, status: Optional[str] = None) -> list[User]:
        """List all users, optionally filtered by status."""
        model = User()
        
        if status:
            # Use status-specific GSI
            model.status = status
            key_condition = model.indexes.get("gsi3").key()
            index_name = "gsi3"
        else:
            # Use general "all users" GSI
            key_condition = model.indexes.get("gsi0").key()
            index_name = "gsi0"
        
        response = self.db.query(
            key=key_condition,
            index_name=index_name,
            table_name=self.table_name
        )
        
        items = response.get("Items", [])
        return [User().map(item) for item in items]
    
    def update_user(self, user_id: str, updates: dict) -> Optional[User]:
        """Update user."""
        user = self.get_user(user_id)
        if not user:
            return None
        
        # Don't allow changing email if it would conflict
        if "email" in updates and updates["email"] != user.email:
            if self.get_user_by_email(updates["email"]):
                raise ValueError("Email already in use")
        
        user.map(updates)
        
        item = user.to_resource_dictionary()
        self.db.save(item=item, table_name=self.table_name)
        
        return user
    
    def deactivate_user(self, user_id: str) -> bool:
        """Soft delete by setting status to inactive."""
        user = self.get_user(user_id)
        if not user:
            return False
        
        user.status = "inactive"
        
        item = user.to_resource_dictionary()
        self.db.save(item=item, table_name=self.table_name)
        
        return True
    
    def search_by_name(self, last_name: str) -> list[User]:
        """Search users by last name."""
        model = User()
        model.last_name = last_name
        
        key_condition = model.indexes.get("gsi1").key(
            condition="begins_with"
        )
        
        response = self.db.query(
            key=key_condition,
            index_name="gsi1",
            table_name=self.table_name
        )
        
        items = response.get("Items", [])
        return [User().map(item) for item in items]
    
    @staticmethod
    def _is_valid_email(email: str) -> bool:
        """Validate email format."""
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return re.match(pattern, email) is not None

Using in Lambda Handlers

Here's how to use your service in a Lambda function:

import json
from your_app.services.user_service import UserService

def lambda_handler(event, context):
    """Create user Lambda handler."""
    service = UserService()
    
    try:
        # Parse request body
        body = json.loads(event.get("body", "{}"))
        
        # Create user
        user = service.create_user(body)
        
        # Return success response
        return {
            "statusCode": 201,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(user.to_dictionary())
        }
    
    except ValueError as e:
        # Business logic error (validation, etc.)
        return {
            "statusCode": 400,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": str(e)})
        }
    
    except Exception as e:
        # Unexpected error
        print(f"Error creating user: {e}")
        return {
            "statusCode": 500,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": "Internal server error"})
        }

Best Practices Summary

✅ Dependency Injection: Allow DynamoDB to be injected for testing
✅ Environment Variables: Use env vars for table names
✅ Return Models: Return typed models, not dictionaries
✅ Business Logic in Services: Keep handlers thin
✅ Single Responsibility: One service per entity/aggregate
✅ Error Handling: Handle DynamoDB exceptions gracefully
✅ Validation: Validate inputs before saving
✅ Documentation: Document what each method does

Testing Services

Because services use dependency injection, they're easy to test:

import unittest
from moto import mock_aws
from your_app.services.user_service import UserService

@mock_aws
class TestUserService(unittest.TestCase):
    def setUp(self):
        # Moto will mock DynamoDB
        self.service = UserService()
        # Create mock table
        # ... setup code ...
    
    def test_create_user(self):
        user = self.service.create_user({
            "id": "test-1",
            "first_name": "John",
            "last_name": "Doe",
            "email": "john@example.com"
        })
        
        self.assertEqual(user.first_name, "John")
        self.assertEqual(user.status, "active")

Conclusion

Service layers are the heart of your application's business logic. They provide:

Clean separation between handlers and database
Testable code through dependency injection
Reusable operations across multiple handlers
Type-safe interfaces using boto3-assist models

Combined with boto3-assist's automatic serialization, Decimal conversion, and index management, you can build maintainable, production-ready applications with confidence.

Get Started

boto3-assist is open source and available on PyPI:

pip install boto3-assist

Check out the complete service layer guide in the documentation.

About the Author: Eric Wilson is the founder of Geek Cafe and created boto3-assist to bring clean architecture patterns to AWS serverless development.

Back to All Blogs