Business Logic Engineering

Rules for "Process Refund":
1. Order must exist and belong to the requesting user.
2. Order must be in "completed" status (not "pending", "cancelled", or already "refunded").
3. Order must have been created within the last 30 days.
4. Refund amount cannot exceed the original order total.
5. If all above pass: order status -> "refunded", payment transaction created, email sent.

This list becomes the source of truth. Tests will verify each rule independently.

Step 2 - Service Layer is Where Rules Live

Business logic belongs exclusively in the service layer. Never in route handlers, never in repositories, never in Pydantic models.

Router: "Received a refund request for order X from user Y"
Service: "Check rules 1-4. Execute rule 5 if all pass. Return result."
Repository: "Write the refund transaction to the DB"

Why this matters: If the same business rule is needed from a webhook handler, a background job, and an admin CLI, the service function is called in all three places. If the rule was in the router, you'd duplicate it three times.

Step 3 - Implement Rules as Explicit Guards

Express each rule as a readable guard clause at the top of the service function:

# services/order_service.py
from datetime import datetime, timedelta, timezone
from uuid import UUID
from fastapi import HTTPException

async def process_refund(order_id: UUID, amount: float, current_user: User, db: AsyncSession) -> RefundOut:
    # Rule 1: Order exists and belongs to user
    order = await order_repo.get_by_id(db, order_id)
    if not order or order.user_id != current_user.id:
        raise HTTPException(status_code=404, detail="Order not found")

    # Rule 2: Order must be in completed status
    if order.status != OrderStatus.COMPLETED:
        raise HTTPException(status_code=422, detail=f"Cannot refund order in '{order.status}' status")

    # Rule 3: Within 30-day window
    refund_window = timedelta(days=30)
    if datetime.now(timezone.utc) - order.created_at > refund_window:
        raise HTTPException(status_code=422, detail="Refund window of 30 days has expired")

    # Rule 4: Amount cannot exceed order total
    if amount > order.total_amount:
        raise HTTPException(status_code=422, detail=f"Refund amount {amount} exceeds order total {order.total_amount}")

    # Rule 5: Execute the refund (all guards passed)
    refund = await payment_service.issue_refund(order, amount, db)
    await order_repo.update_status(db, order_id, OrderStatus.REFUNDED)
    await email_service.send_refund_confirmation(order, refund)

    return RefundOut.model_validate(refund)

Each guard is readable, the reason for rejection is always explicit, and the "happy path" is at the bottom.

Step 4 - State Machine Pattern

When a domain entity has a lifecycle with multiple states and allowed transitions, model it as an explicit state machine:

# models/enums.py
from enum import StrEnum

class OrderStatus(StrEnum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    COMPLETED = "completed"
    REFUNDED = "refunded"
    CANCELLED = "cancelled"

# Define allowed transitions
VALID_TRANSITIONS: dict[OrderStatus, set[OrderStatus]] = {
    OrderStatus.PENDING: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED},
    OrderStatus.CONFIRMED: {OrderStatus.SHIPPED, OrderStatus.CANCELLED},
    OrderStatus.SHIPPED: {OrderStatus.COMPLETED},
    OrderStatus.COMPLETED: {OrderStatus.REFUNDED},
    OrderStatus.REFUNDED: set(),   # terminal
    OrderStatus.CANCELLED: set(),  # terminal
}

# services/order_service.py
def transition_order(order: Order, new_status: OrderStatus) -> None:
    allowed = VALID_TRANSITIONS.get(order.status, set())
    if new_status not in allowed:
        raise HTTPException(
            status_code=422,
            detail=f"Cannot transition order from '{order.status}' to '{new_status}'. Allowed: {[s.value for s in allowed]}"
        )
    order.status = new_status

Never allow arbitrary status changes. Every transition must be explicitly permitted.

Step 5 - Domain Events (Decoupling Side Effects)

When an action triggers multiple side effects (email, analytics, webhook, audit log), decouple them from the core logic using domain events. This keeps the service function focused on the core rule and makes side effects optional and testable independently:

# events/order_events.py
from dataclasses import dataclass
from uuid import UUID

@dataclass
class OrderRefundedEvent:
    order_id: UUID
    user_id: UUID
    refund_amount: float

# In the service, after core logic:
await event_bus.publish(OrderRefundedEvent(
    order_id=order.id,
    user_id=current_user.id,
    refund_amount=amount
))

# Subscribers (registered separately):
event_bus.subscribe(OrderRefundedEvent, send_refund_email_handler)
event_bus.subscribe(OrderRefundedEvent, update_analytics_handler)
event_bus.subscribe(OrderRefundedEvent, notify_webhook_handler)

The service does not know or care about email, analytics, or webhooks. Each handler is independently testable.

Step 6 - Business Rule Unit Tests

Every rule documented in Step 1 gets its own test. The test name maps directly to the rule:

# tests/unit/test_order_service.py

@pytest.mark.asyncio
async def test_refund_fails_if_order_does_not_belong_to_user():
    ...
    with pytest.raises(HTTPException) as exc:
        await process_refund(order_id=other_user_order.id, amount=10.0, current_user=user, db=db)
    assert exc.value.status_code == 404

@pytest.mark.asyncio
async def test_refund_fails_if_order_is_already_refunded():
    ...

@pytest.mark.asyncio
async def test_refund_fails_if_outside_30_day_window():
    order.created_at = datetime.now(timezone.utc) - timedelta(days=31)
    ...

@pytest.mark.asyncio
async def test_refund_succeeds_and_transitions_order_to_refunded_status():
    ...
    assert result.status == "refunded"

Every rule from the plain-language list in Step 1 must have a corresponding failing test and a passing test.

Step 7 - Calculation Logic

Never embed complex calculations inline. Extract them to pure functions that are independently testable:

# services/pricing_service.py

def calculate_order_total(line_items: list[LineItem], discount_code: DiscountCode | None) -> float:
    subtotal = sum(item.unit_price * item.quantity for item in line_items)
    if discount_code and discount_code.discount_type == "percentage":
        discount = subtotal * (discount_code.discount_value / 100)
        subtotal -= discount
    # Tax is calculated on the post-discount subtotal
    tax = subtotal * TAX_RATE
    return round(subtotal + tax, 2)

Pure functions (no DB, no HTTP) with deterministic output are the easiest things to test and the most important to get right.