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 manually run in the PLANTA link modules while developing and debugging.

  • Afterwards most interfaces are called from a macro, running 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.customizer import mail
from ppms import interface

BELASTUNGS_EXPORT = '0f0b2965-0e7a-4ff3-a515-6c8dc5b547fd'
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:])

        subject = 'PLANTA link %s - %s' % (time.strftime('%d.%m.%Y'), subject)

        message = mail.get_default_text_message(recipient=RECIPIENTS_MAIL, subject=subject, content=body)

        with mail.EmailContext() as m:
            response = m.send_message(message=message)        

        return False

    return True
PY

Programmatically creating an interface

Using the PLANTA link API you can create entire interfaces completly 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 something that needs to be validated is changed
  • Sometimes interfaces must change a specific value before running, triggering a sanity check everytime
  • When you know that nobody is going to mess with 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 to provide 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