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 datetime, timedelta
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)
)
today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
yesterday = today - timedelta(days=1)
usages = [
UsageRecord(
usage_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=yesterday.strftime('%Y-%m-%d %H:%M:%S'),
# From when to report.
end_time_utc=today.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()
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()