Examples

Processing Fulfillment Requests

# -*- coding: utf-8 -*-

# This file is part of the Ingram Micro Cloud Blue Connect SDK.
# Copyright (c) 2019 Ingram Micro. All Rights Reserved.

from typing import Union
import warnings

from connect.config import Config
from connect.exceptions import FailRequest, InquireRequest, SkipRequest
from connect.logger import logger
from connect.models import ActivationTemplateResponse, ActivationTileResponse, Fulfillment
from connect.resources import FulfillmentAutomation

# Enable processing of deprecation warnings
warnings.simplefilter('default')

# Set logger level / default level ERROR
logger.setLevel('DEBUG')

# If we remove this line, it is done implicitly
Config(file='config.json')


class FulfillmentExample(FulfillmentAutomation):
    def process_request(self, request):
        # type: (Fulfillment) -> Union[ActivationTemplateResponse, ActivationTileResponse]

        if request.needs_migration():
            # Skip request if it needs migration (migration is performed by an external service)
            logger.info('Skipping request {} because it needs migration.'.format(request.id))
            raise SkipRequest()
        else:
            logger.info('Processing request {} for contract {}, product {}, marketplace {}'
                        .format(request.id,
                                request.contract.id,
                                request.asset.product.name,
                                request.marketplace.name))

            # Custom logic
            if request.type == 'purchase':
                for item in request.asset.items:
                    if item.quantity > 100000:
                        raise FailRequest('Is not possible to purchase product in such quantities')

                for param in request.asset.params:
                    if param.name == 'email' and not param.value:
                        param.value_error = 'Email address has not been provided, ' \
                                            'please provide one'
                        raise InquireRequest(params=[param])

                # Find a param by its id
                param = request.asset.get_param_by_id('purchase_id')
                if param:
                    param.value = '...'  # We can assign the id given by the external service here
                    self.update_parameters(request.id, [param])  # Update param on the platform
                else:
                    raise FailRequest('The asset is expected to have a "purchase_id" param.')

                # Approve by Template
                return ActivationTemplateResponse('TL-497-535-242')
                # Or
                # return TemplateResource().get(pk='TEMPLATE_ID')

                # Approve by ActivationTile
                # return ActivationTileResponse('\n  # Welcome to Fallball!\n\nYes, you decided '
                #                              'to have an account in our amazing service!')
                # Or
                # return TemplateResource().render(pk='TEMPLATE_ID', request_id=request.id)

            elif request.type == 'change':
                # Fail
                raise FailRequest()
            else:
                # Skip request
                raise SkipRequest()


if __name__ == '__main__':
    fulfillment_example = FulfillmentExample()
    fulfillment_example.process()

Processing Tier Config Requests

# -*- coding: utf-8 -*-

# This file is part of the Ingram Micro Cloud Blue Connect SDK.
# Copyright (c) 2019 Ingram Micro. All Rights Reserved.

# NOTE: This example development is in progress. This is just a skeleton.

from typing import Union
import warnings

from connect.config import Config
from connect.logger import logger
from connect.models import ActivationTemplateResponse, ActivationTileResponse, TierConfigRequest
from connect.resources import TierConfigAutomation

# Enable processing of deprecation warnings
warnings.simplefilter('default')

# Set logger level / default level ERROR
logger.setLevel('DEBUG')

# If we remove this line, it is done implicitly
Config(file='config.json')


class TierConfigExample(TierConfigAutomation):
    def process_request(self, request):
        # type: (TierConfigRequest) -> Union[ActivationTemplateResponse, ActivationTileResponse]
        pass


if __name__ == '__main__':
    tier_config_example = TierConfigExample()
    tier_config_example.process()

Reporting Usage Files

# -*- coding: utf-8 -*-

# This file is part of the Ingram Micro Cloud Blue Connect SDK.
# Copyright (c) 2019 Ingram Micro. All Rights Reserved.

from datetime import date, timedelta
import time
import warnings

from connect.config import Config
from connect.logger import logger
from connect.models import Contract, UsageRecord, UsageFile, UsageListing, Product
from connect.resources import UsageAutomation

# Enable processing of deprecation warnings
warnings.simplefilter('default')

# Set logger level / default level ERROR
logger.setLevel('DEBUG')

# If we remove this line, it is done implicitly
Config(file='config.json')


class UsageExample(UsageAutomation):
    def process_request(self, request):
        # type: (UsageListing) -> None

        # Detect specific provider contract
        if request.contract.id == 'CRD-41560-05399-123':
            # Can also be seen from request.provider.id and parametrized further
            # via marketplace available at request.marketplace.id
            usage_file = UsageFile(
                name='sdk test',
                product=Product(id=request.product.id),
                contract=Contract(id=request.contract.id)
            )

            usages = [
                UsageRecord(
                    record_id='unique record value',

                    item_search_criteria='item.mpn',
                    # Possible values are item.mpn or item.local_id.

                    item_search_value='SKUA',
                    # Value defined as MPN on vendor portal.

                    quantity=1,
                    # Quantity to be reported.

                    start_time_utc=(date.today() - timedelta(1)).strftime('%Y-%m-%d'),
                    # From when to report.

                    end_time_utc=time.strftime('%Y-%m-%d %H:%M:%S'),
                    # Till when to report.

                    asset_search_criteria='parameter.param_b',
                    # How to find the asset on Connect.  Typical use case is to use a parameter
                    # provided by vendor, in this case called param_b.  Additionally, asset.id
                    # can be used in case you want to use Connect identifiers.

                    asset_search_value='tenant2'
                )
            ]

            self.submit_usage(usage_file, usages)
        else:
            # Something different could be done here
            pass


if __name__ == '__main__':
    usage_example = UsageExample()
    usage_example.process()

Workflow of Usage Files

# -*- coding: utf-8 -*-

# This file is part of the Ingram Micro Cloud Blue Connect SDK.
# Copyright (c) 2019 Ingram Micro. All Rights Reserved.

import warnings

from connect.config import Config
from connect.logger import logger
from connect.exceptions import AcceptUsageFile, DeleteUsageFile, SkipRequest, SubmitUsageFile
from connect.models import UsageFile
from connect.resources import UsageFileAutomation

# Enable processing of deprecation warnings
warnings.simplefilter('default')

# Set logger level / default level ERROR
logger.setLevel('DEBUG')

# If we remove this line, it is done implicitly
Config(file='config.json')


class UsageFileExample(UsageFileAutomation):
    def process_request(self, request):
        # type: (UsageFile) -> None
        if request.status == 'invalid':
            # Vendor and provider may handle invalid cases differently,
            # probably notifying their staff
            raise DeleteUsageFile('Not needed anymore')
        elif request.status == 'ready':
            # Vendor may submit file to provider
            raise SubmitUsageFile('Ready for provider')
        elif request.status == 'pending':
            # Provider use case, needs to be reviewed and accepted
            raise AcceptUsageFile('File looks good')
        else:
            raise SkipRequest('Non controlled status')


if __name__ == '__main__':
    usage_file_example = UsageFileExample()
    usage_file_example.process()