Data mapping and transformation
Translate any external format into your own models - fields, lists, nested structures and code lookups.
Every external API has its own field names, data structures and conventions. A mapping service translates between them, so each caller receives your consistent model regardless of what the external system returns. This page shows how to map data with data models - individual fields, lists, nested structures and code lookups.
Map fields
When an external API uses different field names than your model - the external system says CarrierCode, your model says airline - the service assigns each field explicitly. The service below calls an airport operations API and returns flights under your own names:
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Model, Service
# External status codes and the values our API returns for them
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):
""" Returns flight data mapped from an external API's field names.
"""
name = 'demo.rest.get-flight-info'
input = FlightInfoRequest
output = FlightInfoResponse
def handle(self) -> 'None':
request:'FlightInfoRequest' = self.request.input
# The external API answers with its own field names ..
conn = self.rest['Airport Ops API']
api_response = conn.get(self.cid, params={'id': request.flight_id})
flight_data = api_response.data
# .. and each field maps to our model explicitly.
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 call it:
The response uses your field names, not the external API's conventions.
Map lists
When an API returns a collection, the service maps each item into a model instance and collects them into a list. The service below fetches all gates in a terminal:
# -*- 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):
name = 'demo.rest.get-terminal-gates'
input = GateListRequest
output = GateListResponse
def handle(self) -> 'None':
request:'GateListRequest' = self.request.input
# The API returns an array of gate objects ..
conn = self.rest['Airport Ops API']
api_response = conn.get(self.cid, params={'terminal': request.terminal_id})
gates_data = api_response.data
# .. each one becomes a Gate instance ..
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)
# .. and the response carries the whole list with its count.
response = GateListResponse()
response.terminal_id = request.terminal_id
response.gates = gates
response.total_count = len(gates)
self.response.payload = response
Flatten nested structures
External APIs often return deeply nested JSON - objects within objects within objects. Long chains like data['customer']['profile']['contact']['email'] are hard to read and debug, so extract each nested section into its own variable and build the flat response from those variables. When a field is missing, the failing line then names exactly which section did not arrive.
The service below flattens a booking that carries passenger data, flight segments and seat assignments 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):
name = 'demo.rest.get-booking-summary'
input = BookingRequest
output = BookingSummary
def handle(self) -> 'None':
request:'BookingRequest' = self.request.input
# The booking arrives as nested JSON ..
conn = self.rest['Reservations API']
api_response = conn.get(self.cid, params={'ref': request.booking_ref})
booking_data = api_response.data
# .. each nested section becomes its own variable ..
passenger = booking_data['passenger']
contact = passenger['contact_info']
flight = booking_data['segments'][0]
departure = flight['departure']
arrival = flight['arrival']
seat = flight['seat_assignment']
# .. and the flat response reads off those 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
Map in both directions
When a service both reads from and writes to an external API, a dedicated mapper class keeps the field name translations in one place - from_external converts external data to your model and to_external converts your model back. When the external API renames a field, you update the mapper and every service that uses it follows:
# -*- 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:'dict') -> 'Passenger':
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:'Passenger') -> 'dict':
return {
'PaxID': passenger.passenger_id,
'GivenName': passenger.first_name,
'FamilyName': passenger.last_name,
'EmailAddress': passenger.email,
'FFN': passenger.frequent_flyer_number
}
class GetPassenger(Service):
name = 'demo.rest.get-passenger'
input = PassengerRequest
output = Passenger
def handle(self) -> 'None':
request:'PassengerRequest' = self.request.input
conn = self.rest['Passenger API']
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):
name = 'demo.rest.update-passenger'
input = UpdatePassengerRequest
output = UpdateResult
def handle(self) -> 'None':
request:'UpdatePassengerRequest' = self.request.input
external_format = PassengerMapper.to_external(request)
conn = self.rest['Passenger API']
conn.put(self.cid, data=external_format)
response = UpdateResult()
response.status = 'updated'
self.response.payload = response
Both services use the same mapper, so the translations stay consistent. To add a middle_name field, you extend the mapper once and both services carry it.
The REST adapter
When many services call the same external system with similar patterns, the REST adapter declares the connection, the HTTP method and the URL as class attributes, and the map_response method carries only the 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):
name = 'adapter.airlines.list'
conn_name = 'Airport Ops API'
method = 'GET'
url_path = '/airlines'
output = AirlineListResponse
def map_response(self, data, **kwargs) -> 'AirlineListResponse':
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 invoke the adapter and receive mapped data, with no knowledge of the external API's field names:
Translate codes to values
External systems use short codes - an aircraft type of 738, a delay reason of WX - where your callers expect readable values like Boeing 737-800 and Weather. Define the translations as module-level dicts, so adding a new code is one line with no logic changes:
# -*- 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):
name = 'demo.rest.get-flight-details'
input = FlightDetailsRequest
output = FlightDetails
def handle(self) -> 'None':
request:'FlightDetailsRequest' = self.request.input
# The API returns short codes ..
conn = self.rest['Airport Ops API']
api_response = conn.get(self.cid, params={'id': request.flight_id})
flight_data = api_response.data
# .. the module-level dicts translate them ..
aircraft_code = flight_data['AircraftCode']
delay_code = flight_data['DelayCode']
# .. and the response carries the readable 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
See also
| Page | What it covers |
|---|---|
| REST adapter | Declarative API calls with map_response and no boilerplate |
| Calling REST APIs | The connections and responses the mapping services work with |
| Orchestration | Combining mapped data from multiple APIs in one response |
| REST channels | The data models that mapping services declare as input and output |