Invoking other services
Zato services can invoke one another synchronously (self.invoke) or in an asynchronous manner (self.invoke_async) so that you can choose whether the invoking service is blocked waiting for the response or if the other service should be run in background.
Note that for the latter, you will not receive a response directly, instead, when invoking another service asynchronously you are given a Correlation ID the service being invoked will also receive so that its response, if any, can be correlated later on, using other Zato mechanisms, specific to your application.
Services can declare their input and output parameters using the input and output attributes. The input attribute accepts either a string tuple of parameter names (with optional - prefix for optional parameters) or a dataclass model. Similarly, output can be a string or a dataclass model. This approach provides type safety, IDE code completion, and automatic validation. Input parameters are accessed via self.request.input and can be simple strings, integers, or complex dataclass models.
For more details on using dataclass models, see Simple IO (SIO).
Synchronous invocation
How to pass a request to another service
You can pass any kind of input to another service, e.g. a dataclass, a plain Python dictionary or any arbitrary Python object.
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self) -> 'None':
# Build the input data for the service we are about to invoke ..
input_data = {'customer_id': 123}
# .. invoke the service ..
response = self.invoke('crm.customer.get-details', input_data)
# .. and log the response.
self.logger.info(response)
Using data formats
Use the 'data_format' parameter if a service expects the incoming data to be in a particular data format, such as JSON or CSV.
# Zato
from zato.common import DATA_FORMAT
from zato.server.service import Service
class MyService(Service):
def handle(self) -> 'None':
# Build the input data ..
input_data = {'order_id': 456, 'status': 'pending'}
# .. store the name of the service to invoke ..
service_name = 'warehouse.order.update-status'
# .. invoke it with JSON data format ..
response = self.invoke(service_name, input_data, data_format=DATA_FORMAT.JSON)
# .. and log the response.
self.logger.info(response)
Inline parameters
Input parameters can be also sent inline, directly in the self.invoke method, just like with a regular Python function call, without creating an input dict or model object upfront:
# Invoke the service with inline parameters
response = self.invoke(service_name, order_id=456, status='pending')
Reading responses
Services will produce their responses and you can use them directly.
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self) -> 'None':
# Build the input data ..
input_data = {'invoice_id': 789, 'customer_id': 123, 'include_items': True}
# .. invoke the service ..
response = self.invoke('billing.invoice.get-details', input_data)
# .. and log the response.
self.logger.info(response)
class GetInvoiceDetails(Service):
# I/O definition - declares which input parameters this service accepts.
# Use '-' prefix for optional parameters, no prefix means required.
input = 'invoice_id', 'customer_id', '-include_items', '-currency_code'
def handle(self) -> 'None':
# Access input parameters via self.request.input ..
invoice_id = self.request.input.invoice_id
customer_id = self.request.input.customer_id
include_items = self.request.input.include_items
# .. business logic here - fetch invoice from database, calculate totals, etc. ..
total_amount = 1500.00
# .. check if items should be included ..
if include_items:
items = [
{'item': 'Product A', 'price': 1000.00},
{'item': 'Product B', 'price': 500.00}
]
else:
items = []
# .. build the response ..
self.response.payload = {
'invoice_id': invoice_id,
'total_amount': total_amount,
'items': items
}
Asynchronous invocation
Background execution
This example shows an asynchronous call, conducted in background:
# Zato
from zato.common import DATA_FORMAT
from zato.server.service import Service
class MyService(Service):
def handle(self) -> 'None':
# Build the input data ..
input_data = {'order_id': 456, 'status': 'pending'}
# .. store the name of the service to invoke ..
service_name = 'warehouse.order.update-status'
# .. invoke it asynchronously with JSON data format ..
cid = self.invoke_async(service_name, input_data, data_format=DATA_FORMAT.JSON)
# .. and log the correlation ID.
self.logger.info(cid)
class UpdateOrderStatus(Service):
def handle(self) -> 'None':
# Note how the payload is a Python dictionary ..
order_id = self.request.payload['order_id']
# .. and log it.
self.logger.info(order_id)
Correlation IDs
When invoking a service asynchronously you do not know when its response will be produced.
Hence, the response from an asynchronous invocation is the Correlation ID the service which is being invoked will receive.
You can use the CID returned to correlate asynchronous responses when they are available - how it is performed is up to your application.
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self) -> 'None':
# Invoke the service asynchronously ..
cid = self.invoke_async('notification.email.send')
# .. and log the correlation ID.
self.logger.info('Got CID %s', cid)
class SendEmailNotification(Service):
def handle(self) -> 'None':
# Log the correlation ID that this service received.
self.logger.info('My CID is %s', self.cid)
Helper API programming tools and techniques
Schedule a meaningful demo
Book a demo with an expert who will help you build meaningful systems that match your ambitions
"For me, Zato Source is the only technology partner to help with operational improvements."
— John Adams
Program Manager of Channel Enablement at Keysight 