Receiving files in a service

A schedule invokes the service you named once for each file that is ready. The file has already been downloaded by the time your service runs, so there is nothing to download and no connection to obtain. The file is kept on the server's local disk and its contents are read only when your service asks for them, which means that a file of any size, including one of many gigabytes, is processed without being loaded into memory.

What the service receives

The item is the service's input:

from zato.server.service import Service

class ProcessInvoice(Service):

    def handle(self):

        item = self.request.input

        self.logger.info('%s from %s, %s bytes', item.file_name, item.conn_name, item.size)

Everything the item contains:

Reading a text file

Most files are text - CSV, fixed-width records, EDIFACT, XML or JSON. Iterating over the item returns the lines of the file one at a time, without their line breaks, and only the current line is held in memory:

from zato.server.service import Service

class ProcessOrders(Service):

    def handle(self):

        for line in self.request.input:
            self.logger.info(line)

When the file is small enough to be handled in one piece, text returns all of it at once:

from zato.server.service import Service

class ProcessInvoice(Service):

    def handle(self):

        text = self.request.input.text
        self.logger.info('Invoice has %s characters', len(text))

The text is read in the encoding selected in the schedule's Encoding field, which is UTF-8 unless a different one is selected. Your service never decodes anything itself. A byte order mark at the start of a file is not part of the text, and Windows line breaks are read the same way as Unix ones, as are the line breaks of EBCDIC files from IBM mainframes.

A file whose bytes are not valid in the schedule's encoding raises UnicodeDecodeError as soon as its text is read, which refuses the file in the same way as any other exception your service raises.

Formats whose records are not separated by line breaks, such as EDIFACT, where each segment ends with an apostrophe, are read with records, which splits the text on the given terminator and returns one record at a time:

from zato.server.service import Service

class ProcessInterchange(Service):

    def handle(self):

        for segment in self.request.input.records("'"):
            self.logger.info(segment)

open_text returns the text as a file object, which is what Python's own readers expect, e.g. csv.reader:

import csv

from zato.server.service import Service

class ImportCustomers(Service):

    def handle(self):

        with self.request.input.open_text() as text_file:
            for row in csv.reader(text_file, delimiter=';'):
                self.logger.info(row)

Reading a binary file

Images, PDF documents, archives and other binary files are read as bytes, without any encoding:

from zato.server.service import Service

class ArchiveScan(Service):

    def handle(self):

        item = self.request.input

        # A PDF document starts with these bytes
        if item.head(5) != b'%PDF-':
            raise Exception(f'Not a PDF document: `{item.full_path}`')

        for chunk in item.chunks():
            self.logger.info('Read %s bytes', len(chunk))

text and data read the whole file into memory each time they are used, so for large files, iterate over the lines, the records or the chunks instead.

The file is deleted from the local disk once your service returns. Anything that has to outlive the invocation is copied out of the item while the service is running.

One file, one invocation

Each file is a separate invocation with its own correlation ID, so a directory holding fifty files produces fifty invocations, each visible on its own in the logs and in the dashboard. Nothing batches them together, and your service never receives a list.

The order of files within one run is the order the remote server listed them in, which is not something to depend on. If files have to be processed in a particular order, that order has to come out of their names or their contents.

Refusing a file

If your service raises, the file is refused. It is left exactly where it was, it is not moved and it is not deleted, and the next run offers it again. Nothing is ever lost because a service failed.

This is what to do with a file you cannot process - validate it, raise on what is wrong, and leave it there until somebody fixes the file or the code:

from zato.server.service import Service

class ProcessInvoice(Service):

    def handle(self):

        item = self.request.input
        text = item.text

        if not text.startswith('INVOICE'):
            raise Exception(f'Not an invoice file: `{item.full_path}`')

        self.logger.info('Taking %s', item.full_path)

A file that keeps being refused keeps coming back, run after run, so a permanently bad file is a permanent error in the logs until it is dealt with. If that is not what you want for a given feed, catch the problem in your own service, put the file's contents somewhere for later inspection, and return normally so the file is moved out of the way.

One refused file has no effect on the others. The run continues to the next file and the healthy ones go through as usual.

Passing the contents on

From here the file is data like any other, and the rest of the platform is available in the usual way - invoke another service, write to a database, publish to a topic, call a REST endpoint:

from zato.server.service import Service

class ProcessInvoice(Service):

    def handle(self):

        item = self.request.input

        # Pass the contents to the service that imports invoices
        self.invoke('billing.import-invoice', {
            'source': item.full_path,
            'received': item.last_modified,
            'data': item.text,
        })

A file that only has to reach another server is uploaded from item.path as it is, without being read at all, whatever its size:

from zato.server.service import Service

class ArchiveInvoice(Service):

    def handle(self):

        item = self.request.input

        conn = self.sftp['Archive SFTP']
        conn.upload(item.path, f'/archive/invoices/{item.file_name}')

See data models for declaring the structure of a file's contents, and data mapping for converting between formats.

A file that is a zip archive, or an IHE XDM package of clinical documents, is read with item.documents() - each file inside the archive is returned as a document with its name, MIME type and bytes, and a file that is not an archive is returned as the one document it is, so a service handles both without telling them apart.

Related

• File transfer schedules - when and how often your service is invoked
• Real-world feeds - the arrangements these services are part of

Learn more