Code Examples

Using the addon_lime_automation.sdk module

As a consultant you can use the module addon_lime_automation.sdk in your solution to easily create automated flow participants connected to a LimeObject.

All you have to do is:

1. Add import of sdk to the top of your file.

import addon_lime_automation.sdk as automation_sdk

2. Create a new instance of the AutomatedFlowParticipantFactory like this:

factory = automation_sdk.AutomatedFlowParticipantFactory(application)

3. Use your factory instance to create automated flow participants. Just supply an automated flow, and the related lime object.

uow = application.unit_of_work()
new_participant, *affected_objects = factory.init_automatedflow_participant(
    automatedflow=my_automatedflow,
    related_object=my_person,
    # "set_to_ready" is default True.
    # Remove it if you want to send it directly to Lime Marketing
    set_to_ready=False,
)

4. Add the objects to your unit of work

uow.add(new_participant)
for affected_object in affected_objects:
    uow.add(affected_object)

5. Set participant to ready and commit your unit of work to send the participant to Lime Marketing.

# Or if it's already set to true in the init function above
new_participant.properties.isready.value = True
uow.commit()

Building an event handler that creates new participants

Create an event handler using lime-project See this guide if you need help.

Use the AutomatedFlowParticipantFactory to create a new automated flow participant when the company changes status from Prospect to Customer.

import addon_lime_automation.sdk as automation_sdk


def company_status_changed(worker, body, message):
    """Summarize your event handlers's functionality here"""
    logger.info("Received message: {}".format(body))
    application = worker.get_application_from_message(message)
    if (
        "buyingstatus" in body["original_values"]
        and "buyingstatus" in body["values"]
    ):
        if body["values"]["buyingstatus"] == "customer":
            logger.info(
                'Company status changed to Customer, \
                add responsible person to "New customers" automated flow'
            )

            uow = application.unit_of_work()
            new_customers_flow = 1001
            company = application.limetypes.company.get(body["values"]["id"])

            factory = automation_sdk.AutomatedFlowParticipantFactory(application)
            # property "isready" is set to True by default,
            # but can be changed using the argument "set_to_ready" below
            # if True, it will automatically be sent to Lime Marketing
            affected_objects = factory.init_automatedflow_participant(
                automatedflow=new_customers_flow,
                related_object=company,
            )
            for affected_object in affected_objects:
                uow.add(affected_object)
            uow.commit()

    message.ack()

Writing tests using automatedflowparticipant or automatedflowproperty

When writing tests where you use the AutomatedFlowParticipant or AutomatedFlowProperty limeobjects you might notice that you get an error similar to this:

.venv\lib\site-packages\mtech_addon_shared\services\configuration_service.py:54: NotFoundError

It is caused by not having a runtime config for addon-lime-automation in the context of the test, so it will not be an issue when running the code on the server. To fix this:

• Create a json-blob of the config and manually set it with lime_data.set_data(). You can find an example config you use here. This can also be done in a fixture.

from lime_data import set_data

cfg = { ... }  # your example config
set_data(self.app, key, json.dumps(cfg), overwrite=True)

Using addon-lime-automation with multiple lime-marketing applications

By default the addon can only be configured to integrate with one lime-marketing application with the API settings made in application_config. There are some cases when the customer uses object access and multiple segments share the same Lime database but have multiple lime-marketing applications. To support this sceanrio the default configuration behavior can be overridden by providing a custom ConfigResolver. In the custom implementation of the resolver it can handle different settings depending on which automated flow participant is in the current context. IE if the participant belongs to this flow which belongs to this group use these lime-marketing api settings.

Warning

The config is subject to change. Use a fixed version of addon-lime-automation in your solution to avoid mismatch in required config for the addon and your custom config provider

Create your own config provider

Theres an abstract class to implement. get_config gets called when the application_config is needed. It takes the limeapplication as an argument and a context dict which holds the id of current 'automatedflowparticipant' if a partcipant is the context - it isnt always so you need to have a fallback config if its not set.

class ConfigProvider(ABC):

    @abstractmethod
    def get_config(self, limeapp: LimeApplication, context: dict):
        pass

Then you have to replace the default config provider with your own

import addon_lime_automation

addon_lime_automation.CONFIG_PROVIDER = CustomConfigProvider()

Example of a custom provider which uses the DefaulConfigProvider as default fallback and checks for the name of the flow in current context. It resolves the config to use from application_config by suffixing the name of the flow

application_name:
    config:
        addon_lime_automation:
            apiurl: https://app.bwz.se/customer_default/bedrock/api
            sweden:
                apiurl: https://app.bwz.se/customer_sweden/bedrock/api
            norway:
                apiurl: https://app.bwz.se/customer_norway/bedrock/api
    secrets:
        addon_lime_automation:
            apikey: apikey_customer_default
            sweden:
                apikey: apikey_customer_sweden
            norway:
                apikey: apikey_customer_norway

import lime_config
import addon_lime_automation
from lime_application import LimeApplication


class CustomConfigProvider(addon_lime_automation.DefaultConfigProvider):
    def __init__(self):
        super().__init__()

    def get_config(self, limeapp: LimeApplication, context: dict):

        id_participant = (context or {}).get('automatedflowparticipant', None)

        logger.info(f"Resolving custom config for particpant {id_participant}")

        cfg_default = super().get_config(limeapp, context)

        try:
            if id_participant is not None:

                participant = limeapp.limetypes.automatedflowparticipant.get(
                    id_participant)
                flow = limeapp.limetypes.automatedflow.get(
                    participant.properties.automatedflow.value)

                # some logic to find out which config to use depending on flow

                config_key = flow.properties.name.value
                logger.info(f"Resolving custom config {config_key}")

                cfg = lime_config.get_app_config(
                    limeapp,
                    f'config.addon_lime_automation.{config_key}')
                cfg_secret = lime_config.get_app_config(
                    limeapp,
                    f'secrets.addon_lime_automation.{config_key}')

                # set overriden endpoint
                if(cfg is not None and cfg_secret is not None):
                    cfg_default['bedrockendpoint']['url'] = cfg.get(
                        'apiurl', None)
                    cfg_default['bedrockendpoint']['apikey'] = cfg_secret.get(
                        'apikey', None)
        except Exception as e:
            logger.error(str(e))

        return cfg_default

# Replace the config provider with our custom provider
addon_lime_automation.CONFIG_PROVIDER = CustomConfigProvider()

Automate the participant creation process

In the addon-lime-automation there exists a number of helper funtions in the sdk module. They are implemented as part of the AutomatedFlowParticipantFactory class.

The functions are:

• create_automatedflow_participant: Will create and commit a participant, returns the participant object.
• init_automatedflow_participant: Will create a participant object and return it, amongs with all related objects.
• get_automated_flow_by_decider: Will fetch and return the correct automated flow based on the lime-admin config.

More details on how the functions work are available in docstrings for each.

An example for how to create a participant depending on a leadstatus for a lead-flow:

import logging
from lime_type import LimeObject
import addon_lime_automation.sdk.decorators as automation_decorators
import addon_lime_automation.sdk as automation_sdk

logger = logging.getLogger(__name__)


@automation_decorators.automated_flow_decider()
class Lead(LimeObject):
    """Summarize the function of a Person object here"""

    def before_update(self, uow, **kwargs):
        """
        This is called on all new and updated objects. All changes
        made to the object here will be persisted.
        All other objects that are changed or created here must be
        added to the unit of work.
        """
        super().before_update(uow, **kwargs)

        create_af_participant(self, uow)


def register_limeobject_classes(register_class):
    register_class("lead", Lead)


def create_af_participant(lead: LimeObject, uow) -> LimeObject:
    leadstatus = lead.properties.leadstatus
    if not leadstatus.is_dirty() or leadstatus.value.key not in [
        "qualified",
        "wip",
    ]:
        return
    factory = automation_sdk.AutomatedFlowParticipantFactory(lead.application)
    flow = lead.properties.automatedflow.fetch()
    person = lead.properties.person.fetch()
    if not flow or not person:
        return

    # Don't forget to add duplicate check if needed.
    new_participant, *affected_objects = factory.init_automatedflow_participant(
        automatedflow=flow,
        related_object=person,
    )
    uow.add(new_participant)
    for affected_object in affected_objects:
        uow.add(affected_object)

    return new_participant