Error handling

Good error handling makes the difference between an API that's easy to integrate with and one that causes frustration. Your services need to return meaningful HTTP status codes and error messages when things go wrong, and gracefully handle failures from external APIs.

This guide covers both directions: returning errors to your API clients, and handling errors from external systems you call.

Returning error responses

When your service can't fulfill a request, set an appropriate HTTP status code and return a JSON error response. Use Python's HTTPStatus enum for readable code.

Setting HTTP status codes

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class GetCustomer(Service):
    name = 'customer.get'

    input = 'customer_id'

    def handle(self):

        customer_id = self.request.input.customer_id

        customer = self.invoke('adapter.crm.customer.get', customer_id=customer_id)

        if not customer:
            self.response.status_code = HTTPStatus.NOT_FOUND
            self.response.payload = {
                'error': 'Customer not found',
                'customer_id': customer_id
            }
            return

        self.response.payload = customer

Test the error case:

# Existing customer - returns 200 with data
curl http://localhost:11223/api/customer/CUST-123

# Non-existent customer - returns 404 with error JSON
curl http://localhost:11223/api/customer/DOES-NOT-EXIST

Common HTTP status codes

Code	Meaning	When to use
200	OK	Successful request
201	Created	Resource created successfully
400	Bad Request	Invalid input data
401	Unauthorized	Missing or invalid credentials
403	Forbidden	Valid credentials but no permission
404	Not Found	Resource doesn't exist
409	Conflict	Resource state conflict
422	Unprocessable Entity	Validation failed
500	Internal Server Error	Unexpected server error
503	Service Unavailable	Temporary unavailability

Validation errors

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class CreateOrder(Service):
    name = 'order.create'

    def handle(self):

        data = self.request.payload
        errors = []

        # Validate required fields
        if not data.get('customer_id'):
            errors.append('customer_id is required')

        if not data.get('items'):
            errors.append('items cannot be empty')

        if data.get('items'):
            for i, item in enumerate(data['items']):
                if not item.get('sku'):
                    errors.append(f'items[{i}].sku is required')
                if not item.get('quantity') or item['quantity'] < 1:
                    errors.append(f'items[{i}].quantity must be positive')

        if errors:
            self.response.status_code = HTTPStatus.BAD_REQUEST
            self.response.payload = {
                'error': 'Validation failed',
                'details': errors
            }
            return

        # Proceed with valid data
        order = self.invoke('adapter.orders.create', data=data)
        self.response.status_code = HTTPStatus.CREATED
        self.response.payload = order

Handling errors from external APIs

Checking response status

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class CallExternalAPI(Service):
    name = 'external.api.call'

    def handle(self):

        conn = self.out.rest['External API'].conn

        response = conn.get(self.cid)

        if response.status_code == HTTPStatus.OK:
            self.response.payload = response.data

        elif response.status_code == HTTPStatus.NOT_FOUND:
            self.response.status_code = HTTPStatus.NOT_FOUND
            self.response.payload = {'error': 'Resource not found in external system'}

        elif response.status_code == HTTPStatus.UNAUTHORIZED:
            self.logger.error('Authentication failed with external API')
            self.response.status_code = HTTPStatus.BAD_GATEWAY
            self.response.payload = {'error': 'External service authentication failed'}

        else:
            self.logger.error(f'External API error: {response.status_code} - {response.data}')
            self.response.status_code = HTTPStatus.BAD_GATEWAY
            self.response.payload = {'error': 'External service error'}

Try/except for API calls

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class SafeAPICall(Service):
    name = 'api.call.safe'

    def handle(self):

        try:
            conn = self.out.rest['Unreliable API'].conn
            response = conn.get(self.cid)
            self.response.payload = response.data

        except Exception as e:
            self.logger.error(f'API call failed: {e}')
            self.response.status_code = HTTPStatus.SERVICE_UNAVAILABLE
            self.response.payload = {
                'error': 'Service temporarily unavailable',
                'retry_after': 60
            }

Partial failures in orchestration

When calling multiple APIs, handle individual failures gracefully:

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class GetDashboard(Service):
    name = 'dashboard.get'

    input = 'user_id'

    def handle(self):

        user_id = self.request.input.user_id
        result = {'user_id': user_id, 'errors': []}

        # Required data - fail if unavailable
        try:
            result['profile'] = self.invoke('adapter.users.get', user_id=user_id)
        except Exception as e:
            self.logger.error(f'Failed to get user profile: {e}')
            self.response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR
            self.response.payload = {'error': 'Failed to load user profile'}
            return

        # Optional data - continue if unavailable
        try:
            result['notifications'] = self.invoke('adapter.notifications.list', user_id=user_id)
        except Exception as e:
            self.logger.warning(f'Notifications unavailable: {e}')
            result['notifications'] = []
            result['errors'].append('notifications_unavailable')

        try:
            result['activity'] = self.invoke('adapter.activity.recent', user_id=user_id)
        except Exception as e:
            self.logger.warning(f'Activity unavailable: {e}')
            result['activity'] = []
            result['errors'].append('activity_unavailable')

        self.response.payload = result

Retry logic

Simple retry with backoff

# -*- coding: utf-8 -*-

# stdlib
import time

# Zato
from zato.server.service import Service

class RetryingAPICall(Service):
    name = 'api.call.with-retry'

    def handle(self):

        max_retries = 3
        retry_delay = 1  # seconds

        conn = self.out.rest['Flaky API'].conn

        for attempt in range(max_retries):
            try:
                response = conn.get(self.cid)

                if response.status_code == 200:
                    self.response.payload = response.data
                    return

                if response.status_code >= 500:
                    # Server error - worth retrying
                    raise Exception(f'Server error: {response.status_code}')

                # Client error - don't retry
                self.response.status_code = response.status_code
                self.response.payload = response.data
                return

            except Exception as e:
                self.logger.warning(f'Attempt {attempt + 1} failed: {e}')

                if attempt < max_retries - 1:
                    time.sleep(retry_delay * (attempt + 1))  # exponential backoff
                else:
                    self.response.status_code = HTTPStatus.SERVICE_UNAVAILABLE
                    self.response.payload = {'error': 'Service unavailable after retries'}

Configuring timeouts

Set timeouts on outgoing connections in Dashboard:

Logging errors

# -*- coding: utf-8 -*-

# stdlib
from http import HTTPStatus

# Zato
from zato.server.service import Service

class LoggingExample(Service):
    name = 'example.logging'

    def handle(self):

        try:
            result = self.do_work()
            self.response.payload = result

        except ValueError as e:
            # Expected error - log as warning
            self.logger.warning(f'Validation error: {e}')
            self.response.status_code = HTTPStatus.BAD_REQUEST
            self.response.payload = {'error': str(e)}

        except Exception as e:
            # Unexpected error - log as error with full context
            self.logger.error(f'Unexpected error processing request', exc_info=True)
            self.response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR
            self.response.payload = {'error': 'Internal server error'}

    def do_work(self):
        return {'status': 'ok'}

Learn more

REST integration overview Start here to see how teams build REST integrations

In-depth API tutorial Master architectural patterns for production-ready integrations

Modern REST API tutorial Design, build and secure REST APIs with Python