interface - Examples
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'):
ppms.ui_message_box('oh oh zum 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
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')
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)
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 theBaseConditional.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)