Every external API has its own field names, data structures, and conventions. Your internal systems have theirs. Data mapping is where these worlds meet - you translate between formats so each system speaks its own language while you maintain a clean, consistent internal model.
This guide shows how to map data using data models, which give you type safety and clean code. For basics on creating channels and handling requests, see REST channels.
External APIs return data in their own formats. Your internal systems expect data in your format. Mapping bridges this gap:
When an external API uses different field names than your internal model, you need to map them explicitly. This is the most common data transformation task - the external system calls a field CarrierCode, but your internal model calls it airline.
The key is to keep the mapping code readable. First, get the data from the external API into a variable. Then, create your response model and populate each field one by one. This makes it clear exactly which external field maps to which internal field.
Here we call an airport operations API that returns flight data with its own naming conventions:
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Model, Service
status_codes = {
'SCH': 'scheduled',
'DEP': 'departed',
'ARR': 'arrived',
'CAN': 'cancelled',
'DLY': 'delayed'
}
@dataclass(init=False)
class FlightInfoRequest(Model):
flight_id: str
@dataclass(init=False)
class FlightInfoResponse(Model):
flight_id: str
airline: str
origin: str
destination: str
scheduled_departure: str
status: str
class GetFlightInfo(Service):
input = FlightInfoRequest
output = FlightInfoResponse
def handle(self):
request:'FlightInfoRequest' = self.request.input
# Get the outgoing connection
conn = self.out.rest['Airport Ops API'].conn
# Call the external API
api_response = conn.get(self.cid, params={'id': request.flight_id})
# Extract the data
flight_data = api_response.data
# Map external field names to our model
response = FlightInfoResponse()
response.flight_id = flight_data['FlightID']
response.airline = flight_data['CarrierCode']
response.origin = flight_data['OriginIATA']
response.destination = flight_data['DestinationIATA']
response.scheduled_departure = flight_data['ScheduledDepartureUTC']
response.status = status_codes[flight_data['StatusCode']]
self.response.payload = response
Create a channel at /api/flights/{flight_id} and test:
The response uses your clean field names, not the external API's conventions. Notice how each step is on its own line - getting the connection, making the call, extracting the data. This makes the code easier to debug and understand.
When an API returns a collection, you need to map each item individually. The pattern is straightforward: iterate through the external data, create a model instance for each item, populate its fields, and collect them into a list.
This example fetches all gates in a terminal. The external API returns an array of gate objects, and we transform each one into our internal Gate model:
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.common.typing_ import list_, optional
from zato.server.service import Model, Service
@dataclass(init=False)
class GateListRequest(Model):
terminal_id: str
@dataclass(init=False)
class Gate(Model):
gate_id: str
gate_number: str
terminal: str
is_international: bool
current_flight: optional[str] = None
@dataclass(init=False)
class GateListResponse(Model):
terminal_id: str
gates: list_[Gate]
total_count: int
class GetTerminalGates(Service):
input = GateListRequest
output = GateListResponse
def handle(self):
request:'GateListRequest' = self.request.input
# Get the connection and call the API
conn = self.out.rest['Airport Ops API'].conn
api_response = conn.get(self.cid, params={'terminal': request.terminal_id})
gates_data = api_response.data
# Map each gate to our model
gates = []
for item in gates_data:
gate = Gate()
gate.gate_id = item['GateID']
gate.gate_number = item['GateNR']
gate.terminal = item['TerminalCode']
gate.is_international = item['ZoneType'] == 'INTL'
gate.current_flight = item['AssignedFlight']
gates.append(gate)
# Build response
response = GateListResponse()
response.terminal_id = request.terminal_id
response.gates = gates
response.total_count = len(gates)
self.response.payload = response
External APIs often return deeply nested JSON - objects within objects within objects. The temptation is to access these with long chains like data['customer']['profile']['contact']['email'], but this makes code hard to read and debug.
Instead, use intermediate variables. Pull out each nested section into its own variable, then access fields from those variables. This makes it obvious where each value comes from, and if something fails, you know exactly which part of the structure was missing.
Here we flatten a booking response that has passenger info, flight segments, and seat assignments all nested inside each other:
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Model, Service
@dataclass(init=False)
class BookingRequest(Model):
booking_ref: str
@dataclass(init=False)
class BookingSummary(Model):
booking_ref: str
passenger_name: str
passenger_email: str
flight_number: str
departure_airport: str
arrival_airport: str
departure_time: str
seat_number: str
cabin_class: str
class GetBookingSummary(Service):
input = BookingRequest
output = BookingSummary
def handle(self):
request:'BookingRequest' = self.request.input
# Call the external API
conn = self.out.rest['Reservations API'].conn
api_response = conn.get(self.cid, params={'ref': request.booking_ref})
booking_data = api_response.data
# Extract nested sections into intermediate variables
passenger = booking_data['passenger']
contact = passenger['contact_info']
flight = booking_data['segments'][0]
departure = flight['departure']
arrival = flight['arrival']
seat = flight['seat_assignment']
# Build flat response from intermediate variables
response = BookingSummary()
response.booking_ref = booking_data['reference']
response.passenger_name = passenger['full_name']
response.passenger_email = contact['email']
response.flight_number = flight['flight_number']
response.departure_airport = departure['airport_code']
response.arrival_airport = arrival['airport_code']
response.departure_time = departure['scheduled_time']
response.seat_number = seat['seat_number']
response.cabin_class = seat['cabin']
self.response.payload = response
Notice how departure and arrival are extracted as separate variables. This avoids repeating flight['departure'] and flight['arrival'] multiple times, and makes the final mapping section clean and readable.
Sometimes you need to both read from and write to an external API. The external system uses its own field names (PaxID, GivenName), and you use yours (passenger_id, first_name). Rather than duplicating the mapping logic in every service, create a dedicated mapper class.
A mapper class has two static methods: from_external converts external data to your model, and to_external converts your model to external format. This keeps all the field name translations in one place. If the external API changes a field name, you only update the mapper.
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Model, Service
@dataclass(init=False)
class PassengerRequest(Model):
passenger_id: str
@dataclass(init=False)
class Passenger(Model):
passenger_id: str
first_name: str
last_name: str
email: str
frequent_flyer_number: str
@dataclass(init=False)
class UpdatePassengerRequest(Model):
passenger_id: str
first_name: str
last_name: str
email: str
frequent_flyer_number: str
@dataclass(init=False)
class UpdateResult(Model):
status: str
class PassengerMapper:
@staticmethod
def from_external(data):
passenger = Passenger()
passenger.passenger_id = data['PaxID']
passenger.first_name = data['GivenName']
passenger.last_name = data['FamilyName']
passenger.email = data['EmailAddress'].lower()
passenger.frequent_flyer_number = data['FFN']
return passenger
@staticmethod
def to_external(passenger):
return {
'PaxID': passenger.passenger_id,
'GivenName': passenger.first_name,
'FamilyName': passenger.last_name,
'EmailAddress': passenger.email,
'FFN': passenger.frequent_flyer_number
}
class GetPassenger(Service):
input = PassengerRequest
output = Passenger
def handle(self):
request:'PassengerRequest' = self.request.input
conn = self.out.rest['Passenger API'].conn
api_response = conn.get(self.cid, params={'id': request.passenger_id})
passenger_data = api_response.data
self.response.payload = PassengerMapper.from_external(passenger_data)
class UpdatePassenger(Service):
input = UpdatePassengerRequest
output = UpdateResult
def handle(self):
request:'UpdatePassengerRequest' = self.request.input
external_format = PassengerMapper.to_external(request)
conn = self.out.rest['Passenger API'].conn
conn.put(self.cid, data=external_format)
response = UpdateResult()
response.status = 'updated'
self.response.payload = response
Both services use the same mapper, so the field name translations are consistent. If you later add a middle_name field, you add it to the mapper once and both services automatically handle it.
When you have many similar API integrations - all calling the same external system with similar patterns - the RESTAdapter provides a structured approach. Instead of writing the connection and call logic in every service, you define it once in the adapter class.
The adapter has a map_response method where you transform the external data to your model. This is called automatically after the API call completes. The adapter handles the connection, HTTP method, and URL - you just focus on the data transformation.
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.common.typing_ import list_
from zato.server.service import Model, RESTAdapter
@dataclass(init=False)
class Airline(Model):
code: str
name: str
country: str
@dataclass(init=False)
class AirlineListResponse(Model):
airlines: list_[Airline]
class GetAirlines(RESTAdapter):
conn_name = 'Airport Ops API'
method = 'GET'
url_path = '/airlines'
output = AirlineListResponse
def map_response(self, data, **kwargs):
airlines = []
for item in data:
airline = Airline()
airline.code = item['IATA']
airline.name = item['AirlineName']
airline.country = item['CountryCode']
airlines.append(airline)
response = AirlineListResponse()
response.airlines = airlines
return response
Other services can now call this adapter to get clean, mapped data. They don't need to know anything about the external API's field names or structure:
External systems often use short codes that are meaningless to your users. An aircraft type might be 738, a delay reason might be WX. Your API should return human-readable values like Boeing 737-800 and Weather.
Define these translations as simple dicts at the module level. This makes them easy to find and update. When you need to add a new code, you just add a line to the dict - no logic changes needed.
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Model, Service
aircraft_types = {
'738': 'Boeing 737-800',
'763': 'Boeing 767-300',
'388': 'Airbus A380-800',
'320': 'Airbus A320',
'321': 'Airbus A321'
}
delay_reasons = {
'WX': 'Weather',
'TC': 'Technical',
'OP': 'Operational',
'SC': 'Security',
'CR': 'Crew'
}
@dataclass(init=False)
class FlightDetailsRequest(Model):
flight_id: str
@dataclass(init=False)
class FlightDetails(Model):
flight_id: str
aircraft_type: str
delay_reason: str
class GetFlightDetails(Service):
input = FlightDetailsRequest
output = FlightDetails
def handle(self):
request:'FlightDetailsRequest' = self.request.input
# Call the API
conn = self.out.rest['Airport Ops API'].conn
api_response = conn.get(self.cid, params={'id': request.flight_id})
flight_data = api_response.data
# Look up the codes
aircraft_code = flight_data['AircraftCode']
delay_code = flight_data['DelayCode']
# Build response with translated values
response = FlightDetails()
response.flight_id = flight_data['FlightID']
response.aircraft_type = aircraft_types[aircraft_code]
response.delay_reason = delay_reasons[delay_code]
self.response.payload = response