Die nachfolgenden Inhalte sind nur in englischer Sprache verfügbar.

This topic contains various examples on how to achieve different things using the PLANTA link API.

Executing an interface from code

  Most interfaces are only run manually in the PLANTA link modules for development and debugging purposes.

  • Later, most interfaces are called from a macro which runs when the user performs a specific action or based on a timed job on the server.
  • This example shows you how to initialize a template, copy it, execute it, check the results and send a mail when something goes wrong
# -*- coding: utf-8 -*-
import time

from ppms import ppms
from ppms import ppms_cu
from ppms import interface

BELASTUNGS_EXPORT = '0f0b2965-0e7a-4ff3-a515-6c8dc5b547fd'

NOTIFICATION_MAIL_ADRESS = 'interface@company.com'
RECIPIENTS_MAIL = 'support@company.com'


def on_load():
    # We need a module instance to copy the config
    mod = ppms.get_macro_module()

    # Initialize the template config we want to use. Use a StaticConfig for better performance
    template_config = interface.StaticConfig(config_id=BELASTUNGS_EXPORT)
    if template_config is None:
        raise ValueError('No config with id "%s"' % BELASTUNGS_EXPORT)

    # Since templates can't be executed we need to make a copy before executing the interface. Since our parent is a StaticConfig our copy will also be one!
    config = template_config.copy(invoker_module=mod)
    config_id = config.config_id

    # Change the name to be able to identify this config more easily later
    new_description = 'Nachtlauf %s' % time.strftime('%d.%m.%Y')
    config.description = new_description

    # Load the data into the pool
    counter = interface.transfer_step_one(invoker_module=mod, config=config)
    if not check_results(config, counter, 'Source -> Pool'):
        return

    # Load the data into the target
    counter = interface.transfer_step_two(invoker_module=mod, config=config)
    check_results(config, counter, 'Pool -> Target')
    
    # If we didn't have a pool we would call direct_transfer like this:
    #counter = interface.direct_transfer(invoker_module=mod, config=config)
    #check_results(config, counter, 'Datensätze ins Ziel übertragen')

# This function checks the results from a transfer and sends a mail when errors were encountered
def check_results(config, counter, subject):
    sent_records = counter.sent_records
    received_records = counter.received_records
    failed_records = counter.errors
    critical = counter.critical

    body = 'PLANTA <em>link</em> Report\n\n'

    if critical:
        body = 'Critical Error during transaction!\n'

    if failed_records:
        body += 'Sent: {sent_records}\n\n' \
                'Succeeded: {received_records}\n' \
                'Failed: {failed_records}'
        body = body.format(sent_records=sent_records, received_records=received_records, failed_records=failed_records)

    if critical or failed_records:
        # This only works if you're logging to PLANTA and not a file
        body += '\n\n' \
                'Last 20 rows from the log:\n\n'

        body += '\n'.join(config.log_content.split('\n')[-20:])

        ppms_cu.Action.send_email(my_email=NOTIFICATION_MAIL_ADRESS,
                                  recipients_email=RECIPIENTS_MAIL,
                                  subject='PLANTA <em>link</em> %s - %s' % (time.strftime('%d.%m.%Y'), subject),
                                  email_body=body)

        return False

    return True
PY

Programmatically creating an interface

Using the PLANTA link API you can create complete interfaces entirely from code.

  • This is most useful when writing unittests
  • The mapping we will create will look like this:

from ppms.interface import Config, MappingType

# We aren't using StaticConfig because we want to modify the structure at runtime
config = Config.create(description='Example Interface', template=True)

# Create the first source mapping
source = config.create_mapping(type=MappingType.SOURCE, object='module_id')

# Create a validator as a child to this source mapping
validator = source.create_child(type=MappingType.VALIDATOR, object='ExistsAsPK')

# Modify the parameters
validator.modify_parameter('table_num', '405')
validator.modify_parameter('child_when_invalid', 'ConstantValue')

# Create a target mapping and enricher as children to the validator
validator_target = validator.create_child(type=MappingType.TARGET, object='module_id')
enricher = validator.create_child(type=MappingType.ENRICHER, object='ConstantValue')

# Create another target mapping as a child to the enricher
enricher_target = enricher.create_child(type=MappingType.TARGET, object='constant')
PY

Disabling the validation for an interface

The validation step can be quite tedious when you have a complex interface.

  • Luckily, PLANTA link will cache the result and only revalidate the configuration when changes are made that need to be validated
  • Sometimes interfaces must change a specific value before they can run, triggering a sanity check each time
  • When you know that nobody is going to alter the template and invalidate it, you can skip the validation before execution by overriding the is_valid property
from ppms.interface import StaticConfig, direct_transfer

BELASTUNGS_EXPORT = '0f0b2965-0e7a-4ff3-a515-6c8dc5b547fd'


class ValidatedStaticConfig(StaticConfig):
    
    # Overriding is_valid to always return True tells PLANTA <em>link</em> that the validation always succeeds
    @property
    def is_valid(self):
        return True
        
mod = ppms.get_macro_module()

template_config = ValidatedStaticConfig(config_id=BELASTUNGS_EXPORT)
config = template_config.copy(invoker_module=mod)

direct_transfer(invoker_module=mod, config=config)
PY

Implementing a new conditional

You can write your own classes that derive from ppms.interface.BaseConditional, providing new conditions to determine which parameters PLANTA link should use

  • All you need to do is write a class that inherits from ppms.interface.BaseConditional and implement the BaseConditional.value property
from ppms import ppms
from ppms.interface import BaseConditional
from ppms.constants import SYS_VAR_LOGGED_IN_USER


# This conditional will return the current user
# You could use this f.e. to configure that only a certain user
# may execute a interface on production
class UserConditional(BaseConditional):
    
    @property
    def value(self):
        return ppms.uvar_get(SYS_VAR_LOGGED_IN_USER)
PY