Data Exchange Service

The data exchange service is a separate module to formalize data handling in terms of logging and locking.

Add backend components

In order to deploy the model and services from the framework to your project, you have to import them both. To do so, add the following two packages including their content into their respective paths.

model/valuea/des/__init__.py

from valuea_framework.model.valuea.des import *

services/valuea/des/__init__.py

import valuea_framework.services.valuea.des
import pkgutil
__path__ = pkgutil.extend_path(valuea_framework.services.valuea.des.__path__,
                               valuea_framework.services.valuea.des.__name__)

The first one makes sort deploymodel --rdbms will known which tables to create, the other one imports our standard services into the valuea.des namespace.

Management user interface

Part of the framework is a standard module which utilizes the services installed earlier for management purposes, when this module is installed you can easily inspect the jobs started by the system and disable future executions if needed.

To install, use the following command on a frontend server:

pkg add -f os-valuea-des-x.y.txz

usage

The basic concept of DES is to facilitate a standard import workflow for files and messages, which in many cases should be handled as a singleton operation. DES relies on a model and uses standard database locking to assure consistency among difference processes.

Imagine a set files needs to be processed, but only when they are in a consistent state on disk and we need to assure that another process trying to do the same won’t start in the same timeframe. Our process might look as follows.

We first would register our task which identifies our job, then check if we are ready to run, if so wait for the files to complete , run our import action (whatever it looks like) and close our process.

In code, without the actual import with logging included this would looks as follows (__file__ is the script itself).

des_example.py

import valuea_framework.des

desjob = valuea_framework.des.DESJob('XXX')
if desjob.lock() and desjob.is_enabled():
    desjob.create_instance()
    desjob.log('wait for files', log_type="INFO", flush=True)
    valuea_framework.des.wait_for_files([__file__])
    desjob.log('start import', log_type="INFO", flush=True)
    desjob.log('do run() action', log_type="INFO", flush=True)
    desjob.close_instance(u'PCD')
    # desjob.disable()
    desjob.unlock()

The log information can be examined using the management console installed in the previous chapter.