Invoking other services

Synchronous and asynchronous invocation of one service from another, with correlation IDs.

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 data models.

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
        }

What self.invoke returns

The return value is the target service's response payload itself, not a wrapper object. What that is depends on what the target assigned to self.response.payload:

The target service set its payload to	`self.invoke` returns
A dict, list, string or bytes	The same object, unchanged
A data model instance	The model instance itself, still a model
Fields assigned one by one, e.g. `self.response.payload.abc = 1`	A plain dict of those fields
Nothing at all	An empty string

Two of these surprise people, so they are worth spelling out:

A model stays a model across self.invoke - it is converted to a dict only when it crosses a REST channel, so the caller can use response.abc dot access and pass the object onward with its typing intact
A service that produced no response yields an empty string, never None and never an empty dict, so a "did I get anything" check is if response:

Useful keyword arguments:

# Kill the invocation if it runs longer than 10 seconds -
# a timeout raises an exception in the calling service
response = self.invoke('crm.customer.get-details', input_data, timeout=10)

# Receive dot-access Bunch objects instead of plain dicts
# for payloads that were built field by field
response = self.invoke('crm.customer.get-details', input_data, as_bunch=True)

The first argument can also be the service's class instead of its name, which keeps refactoring safe - renaming the service in one place cannot break its callers:

from api.crm import GetCustomerDetails

response = self.invoke(GetCustomerDetails, input_data)

Invoking a service that is inactive raises an Inactive exception, and a service cannot invoke itself.

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(f'Got CID {cid}')

class SendEmailNotification(Service):
    def handle(self) -> 'None':

        # Log the correlation ID that this service received.
        self.logger.info(f'My CID is {self.cid}')

INFO - Got CID c37780e779666e59ed284f92
INFO - My CID is c37780e779666e59ed284f92