Die Dokumentation ab Version 39.5.17 von PLANTA project finden Sie in der neuen PLANTA Online-Hilfe.

interface - Examples

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 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 Link 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 Link %s - %s' % (time.strftime('%d.%m.%Y'), subject),
                                  email_body=body)

        return False

    return True

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:
example_mapping.png

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')

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 Link 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)

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)
Topic attachments
ISorted ascending Attachment History Size Date Comment
Pngpng example_mapping.png r1 19.4 K 2017-10-10 - 13:31  

         PLANTA project









 
  • Suche in Topic-Namen

  • Suche in Topic-Inhalten
This site is powered by the TWiki collaboration platform Powered by Perl