interface - Examples

# -*- 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

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

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)

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)