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

Executing a 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 a 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