Be sure to read the modern REST API tutorial in Python once you've finished this chapter.
The REST adapter is a utility base class that simplifies calling external REST APIs. By subclassing RESTAdapter, you can invoke REST endpoints declaratively without implementing the usual self.handle method.
The adapter handles all the low-level details of making HTTP requests, including connection management, authentication, request serialization, response parsing, and automatic retry logic. You simply declare what connection to use and how to map the data, and the adapter takes care of the rest.
To create a REST adapter service, subclass RESTAdapter and set the conn_name attribute to the name of an outgoing REST connection configured in your Zato environment. This is the only required attribute. The adapter will automatically invoke the configured endpoint when the service runs, and the response will be available as self.response.payload.
By default, the adapter uses the GET method. If you need to use POST, PUT, DELETE, or another HTTP method, set the method attribute accordingly.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class GetFlight(RESTAdapter):
""" Retrieves flight data from the scheduling system.
"""
conn_name = 'flight.ops'
To use a different HTTP method:
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class UpdateRunwayStatus(RESTAdapter):
""" Updates runway availability status.
"""
conn_name = 'runway.ops'
method = 'POST'
Once you have defined your adapter service, you have two main ways to use it:
Option 1: Invoke from another service - Call the adapter from a higher-level service using self.invoke. An orchestration service is simply a service that coordinates calls to multiple other services or APIs to accomplish a business task.
For example, to display a flight's full status, you might need to call one API for flight details, another for weather at the destination, and a third for gate information. The orchestration service calls each adapter, combines the results, and returns a unified response.
This separation keeps each adapter focused on one external system while the orchestration service handles the business logic.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetFlightStatus(Service):
def handle(self):
flight = self.invoke('airport.adapter.get-flight', id=self.request.input.flight_id)
self.response.payload = flight
Option 2: Expose directly via REST channel - Mount the adapter service on a REST channel to create a simple API gateway. External clients call your Zato endpoint, and the adapter forwards the request to the external system, transforms the response, and returns it. This requires no additional code beyond the adapter class itself.
In many real-world scenarios, you need to determine request parameters at runtime based on input data, configuration, or other factors. The REST adapter supports this through a set of get_* methods that you can override.
When the adapter runs, it checks for these methods and calls them to obtain dynamic values. This is useful when you need to set headers based on the current request, add query parameters conditionally, or determine the HTTP method based on input. Each method receives the current state and returns the values to use for that particular invocation.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class GetWeatherData(RESTAdapter):
""" Fetches weather data for airport operations.
"""
conn_name = 'weather.api'
def get_method(self):
return self.request.input.get('method', 'GET')
def get_headers(self):
return {
'X-Request-ID': self.cid,
'X-Airport-Code': self.request.input.get('airport_code', 'BIKF'),
}
def get_query_string(self, params):
params['unit'] = 'metric'
return params
When calling external APIs, the response format is often different from what your application expects. External systems may use different field names, nested structures, or legacy formats that don't match your canonical data model.
The REST adapter integrates with Zato's Model system to automatically deserialize JSON responses into Python dataclass objects. Simply set the model attribute to your Model class, and the adapter will parse the response and create an instance of that model, giving you type-safe access to the response data.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Model, RESTAdapter
class AircraftModel(Model):
id: int
registration: str
type_code: str
class GetAircraft(RESTAdapter):
conn_name = 'fleet.api'
model = AircraftModel
For more complex transformations, implement a map_response method. This method receives the parsed response data (already converted to your model class if specified) and returns your canonical data model.
This pattern is essential when integrating with external systems that use different field names, codes, or formats than your internal systems. You can look up values in configuration, convert between formats, filter data, or perform any other transformation needed to produce a clean, consistent output for the rest of your application.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Model, RESTAdapter
class ExternalFlightData(Model):
flight_id: int
carrier: str
dep_time_utc: str
status_code: str
class Flight(Model):
id: int
airline: str
departure_time: str
status: str
class GetFlightDetails(RESTAdapter):
input = 'id'
output = Flight
has_query_string_id = True
conn_name = 'Flight.Schedule'
model = ExternalFlightData
def map_response(self, data:'ExternalFlightData') -> 'Flight':
status = self.config.airport.status_codes[data.status_code]
airline = self.config.airport.carriers[data.carrier]
out = Flight()
out.id = data.flight_id
out.airline = airline
out.departure_time = data.dep_time_utc
out.status = status
return out
Many REST APIs use path parameters to identify resources, such as /terminals/{terminal_id}/gates/{gate_id}. When your outgoing connection URL contains placeholders like {terminal_id}, you need to provide values for these parameters at runtime.
Implement the get_path_params method to return a dictionary of parameter names and values. The adapter will substitute these values into the URL before making the request. Any parameters not used in the URL path will be passed as query string parameters instead.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class GetGateAssignment(RESTAdapter):
input = 'terminal_id', 'gate_id'
conn_name = 'Gate.Management'
method = 'GET'
def get_path_params(self, params):
return {
'terminal_id': self.request.input.terminal_id,
'gate_id': self.request.input.gate_id,
}
For POST, PUT, and PATCH requests, you typically need to send data in the request body. Implement the get_request method to return the data that should be sent. The adapter will automatically serialize dictionaries and Model objects to JSON.
This method gives you full control over the request payload. You can build complex nested structures, transform input data, merge data from multiple sources, or apply any business logic needed before sending the request to the external system.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class SubmitCargoManifest(RESTAdapter):
input = 'manifest'
conn_name = 'Cargo.System'
method = 'POST'
def get_request(self):
return self.request.input.manifest
When integrating with an external system, you often need multiple adapter services that share common configuration such as authentication, base URLs, or headers. Instead of repeating this configuration in every service, create a base adapter class that other services inherit from.
This pattern keeps your code DRY and makes it easy to update shared configuration in one place. For example, if an external system changes its authentication method, you only need to update the base adapter class, and all services that inherit from it will automatically use the new configuration.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import RESTAdapter
class BaseAirportAdapter(RESTAdapter):
sec_def_name = 'Airport.OAuth.Token'
class GetRunway(BaseAirportAdapter):
conn_name = 'Airfield.Runways'
input = 'id'
has_query_string_id = True
class GetTaxiway(BaseAirportAdapter):
conn_name = 'Airfield.Taxiways'
input = 'id'
has_query_string_id = True
REST adapter services are regular Zato services, which means you can invoke them from other services using self.invoke, just like any other service. This makes it easy to build higher-level orchestration services that coordinate multiple API calls.
The adapter service receives input parameters through the standard self.request.input mechanism, calls the external API, transforms the response, and returns the result. The calling service receives this result directly from self.invoke, making it seamless to chain multiple API calls together.
Adapter services can also be exposed via REST channels, allowing external clients to call them directly. This is useful when you want to provide a simplified or transformed view of an external API to your own clients.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class SyncGateAssignments(Service):
def handle(self):
flight_list = self.invoke('airport.adapter.get-flight-list')
gate_list = self.invoke('airport.adapter.get-gate-list', status='available')
for flight in flight_list:
if flight.gate_id not in gate_list:
self.invoke('airport.adapter.assign-gate', flight_id=flight.id)
| Name | Default | Description |
|---|---|---|
| conn_name | '' | Name of the outgoing REST connection to use |
| method | 'GET' | HTTP method (GET, POST, PUT, DELETE, PATCH, etc.) |
| get_conn_name | None | Callable returning the connection name dynamically |
| get_method | None | Callable returning the HTTP method dynamically |
| Name | Default | Description |
|---|---|---|
| get_request | None | Callable returning the request body data |
| get_query_string | None | Callable returning query string parameters as a dict |
| get_path_params | None | Callable returning path parameters as a dict |
| get_headers | None | Callable returning HTTP headers as a dict |
| has_query_string_id | False | If True, automatically adds an 'id' parameter to query string from input |
| query_string_id_param | None | Custom name for the query string ID parameter (default: 'id') |
| has_json_id | False | If True, uses JSON ID parameter |
| json_id_param | None | Custom name for the JSON ID parameter |
| Name | Default | Description |
|---|---|---|
| sec_def_name | None | Name of the security definition to use |
| auth_scopes | '' | OAuth scopes to request |
| get_sec_def_name | None | Callable returning the security definition name dynamically |
| get_auth_scopes | None | Callable returning auth scopes dynamically |
| get_auth_bearer | None | Callable returning a bearer token to add to the Authorization header |
| Name | Default | Description |
|---|---|---|
| model | None | A Model class to map the response to |
| map_response | None | Callable to transform the response data |
| log_response | False | Whether to log the response |
| needs_raw_response | False | If True, returns a RESTAdapterResponse with both parsed data and raw response |
| Name | Default | Description |
|---|---|---|
| max_retries | 0 | Maximum number of retry attempts on failure |
| retry_sleep_time | 2 | Seconds to wait between retries |
| retry_backoff_threshold | 3 | Number of retries before applying backoff multiplier |
| retry_backoff_multiplier | 2 | Multiplier applied to sleep time after threshold |