Develop

Create an events application class

Define a python module that includes your application type. The following showcases how to create a module that implements an events application:

$ touch tshirt_extension/eventsapp.py

Open tshirt_extension/eventsapp.py with your code editor and provide the following specifications:

from connect.eaas.core.extension import EventsApplicationBase


class TShirtEventsApplication(EventsApplicationBase):
    pass

Declare events application class in pyproject.toml

Your extension will be executed by the EaaS runtime called Connect Extension Runner.

The extension runner utilizes setup tools entrypoints to discover all features that are provided by your extension. Consequently, it is required to modify pyproject.toml and add an entrypoint for your events application as follows:

[tool.poetry]
name = "tshirt-extension"
version = "0.1.0"
description = "T-Shirt extension"
authors = ["Globex corporation"]
readme = "README.md"
packages = [{include = "tshirt_extension"}]

[tool.poetry.dependencies]
python = ">=3.8,<3.11"
connect-eaas-core = ">=26.12,<27"

[tool.poetry.plugins."connect.eaas.ext"]
"eventsapp" = "tshirt_extension.eventsapp:TShirtEventsApplication"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Track out of stock items

Track items that are out of stock by declaring corresponding environment variables. The following showcases how to create a variable that will be automatically created and assigned to its initial value once the extension is launched for the first time:

from connect.eaas.core.decorators import variables
from connect.eaas.core.extension import EventsApplicationBase


@variables(
    [
        {
            'name': 'OUT_OF_STOCK_SIZE',
            'initial_value': 'm',
        },
    ],
)
class TShirtEventsApplication(EventsApplicationBase):
    pass

Update the OUT_OF_STOCK_SIZE variable

Once your variable for tracking out of stock items is defined, create a schedule to automatically update this variable. The following demonstrates how to create a method that will run and check for updates once a day. This method will get a random size from the list and update the OUT_OF_STOCK_SIZE variable:

import random

from connect.eaas.core.decorators import schedulable, variables
from connect.eaas.core.extension import EventsApplicationBase
from connect.eaas.core.responses import ScheduledExecutionResponse


TSHIRT_SIZES = ['xs', 's', 'm', 'l', 'xl']


@variables(
    [
        {
            'name': 'OUT_OF_STOCK_SIZE',
            'initial_value': 'm',
        },
    ],
)
class TShirtEventsApplication(EventsApplicationBase):
    @schedulable(
        'Update stock',
        'Set the OUT_OF_STOCK_SIZE variable with a random size that will be considered out of stock',
    )
    def update_stock(self, schedule):
        current_oos_size = random.choice(TSHIRT_SIZES)
        self.client.ns(
            'devops',
        ).collection(
            'services',
        ).resource(
            '',
        ).collection(
            'environments',
        ).resource(
            '',
        ).collection(
            'variables',
        ).resource(
            'OUT_OF_STOCK_SIZE',
        ).update(
            payload={'value': current_oos_size},
        )
        return ScheduledExecutionResponse.done()

Method for draft validation events

The following showcases how to create an event handler for the draft purchase requests validation. This handler will check whether the selected t-shirt size is out of stock. To implement the draft request validation, add the asset_purchase_request_validation event type and specify the draft status:

import random

from connect.eaas.core.decorators import event, schedulable, variables
from connect.eaas.core.extension import EventsApplicationBase
from connect.eaas.core.responses import InteractiveResponse, ScheduledExecutionResponse

TSHIRT_SIZES = ['xs', 's', 'm', 'l', 'xl']


@variables(
    [
        {
            'name': 'OUT_OF_STOCK_SIZE',
            'initial_value': 'm',
        },
    ],
)
class TShirtEventsApplication(EventsApplicationBase):
    @schedulable(
        'Update stock',
        'Set the OUT_OF_STOCK_SIZE variable with a random size that will be considered out of stock',
    )
    def update_stock(self, schedule):
        current_oos_size = random.choice(TSHIRT_SIZES)
        self.client.ns(
            'devops',
        ).collection(
            'services',
        ).resource(
            '',
        ).collection(
            'environments',
        ).resource(
            '',
        ).collection(
            'variables',
        ).resource(
            'OUT_OF_STOCK_SIZE',
        ).update(
            payload={'value': current_oos_size},
        )
        return ScheduledExecutionResponse.done()

    @event('asset_purchase_request_validation')
    def validate_tshirt_size(self, request):
        if request['asset']['params'][0]['value'] == self.config['OUT_OF_STOCK_SIZE']:
            request['asset']['params'][0]['value_error'] = (
                f"Sorry, the size {self.config['OUT_OF_STOCK_SIZE']} is out of stock"
            )
        else:
            request['asset']['params'][0]['value_error'] = ''

        return InteractiveResponse.done(body=request)

Method to auto-approve pending requests

Add an event handler to automatically approve pending purchase requests. Specifically, add the asset_purchase_request_processing event type and the pending state:

import random

from connect.eaas.core.decorators import event, schedulable, variables
from connect.eaas.core.extension import EventsApplicationBase
from connect.eaas.core.responses import (
    BackgroundResponse,
    InteractiveResponse,
    ScheduledExecutionResponse,
)

TSHIRT_SIZES = ['xs', 's', 'm', 'l', 'xl']


@variables(
    [
        {
            'name': 'OUT_OF_STOCK_SIZE',
            'initial_value': 'm',
        },
    ],
)
class TShirtEventsApplication(EventsApplicationBase):
    @schedulable(
        'Update stock',
        'Set the OUT_OF_STOCK_SIZE variable with a random size that will be considered out of stock',
    )
    def update_stock(self, schedule):
        current_oos_size = random.choice(TSHIRT_SIZES)
        self.client.ns(
            'devops',
        ).collection(
            'services',
        ).resource(
            '',
        ).collection(
            'environments',
        ).resource(
            '',
        ).collection(
            'variables',
        ).resource(
            'OUT_OF_STOCK_SIZE',
        ).update(
            payload={'value': current_oos_size},
        )
        return ScheduledExecutionResponse.done()

    @event('asset_purchase_request_validation', statuses=['draft'])
    def validate_tshirt_size(self, request):
        if request['asset']['params'][0]['value'] == self.config['OUT_OF_STOCK_SIZE']:
            request['asset']['params'][0]['value_error'] = f"Sorry, the size {self.config['OUT_OF_STOCK_SIZE']} is out of stock"

        return InteractiveResponse.done(body=request)

    @event('asset_purchase_request_processing', statuses=['pending'])
    def approve_purchase_request(self, request):
        product_id = request['asset']['product']['id']
        template = self.client.products[product_id].templates.filter(
            scope='asset',
            type='fulfillment',
        ).first()

        self.client.requests[request['id']]('approve').post(
            {'template_id': template['id']},
        )
        return BackgroundResponse.done()

Congratulations

This concludes your Events Application creation and your app should be ready for the following tests!