Merge pull request #96 from cloudblue/lite-28889-billing-request-lookup-transformation

Add billing request transformation

Author: bdjilka

connect_transformations/lookup_billing_request/__init__.py

@@ -0,0 +1,5 @@
+from connect_transformations.exceptions import BaseTransformationException
+
+
+class BillingRequestLookupError(BaseTransformationException):
+    """Exception that happens when the billing request lookup fails"""

connect_transformations/lookup_billing_request/exceptions.py

@@ -0,0 +1,187 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2023, CloudBlue LLC
+# All rights reserved.
+#
+from typing import Dict
+
+from connect.client import AsyncConnectClient
+from connect.eaas.core.decorators import router, transformation
+from connect.eaas.core.inject.asynchronous import get_installation_client
+from connect.eaas.core.responses import RowTransformationResponse
+from fastapi import Depends
+
+from connect_transformations.constants import SEPARATOR
+from connect_transformations.lookup_billing_request.exceptions import BillingRequestLookupError
+from connect_transformations.lookup_billing_request.models import (
+    Configuration,
+    SubscriptionParameter,
+)
+from connect_transformations.lookup_billing_request.utils import validate_lookup_billing_request
+from connect_transformations.models import Error, ValidationResult
+from connect_transformations.utils import deep_itemgetter, is_input_column_nullable
+
+
+class LookupBillingRequestTransformationMixin:
+    @transformation(
+        name='Lookup CloudBlue Billing request data',
+        description=(
+            'This transformation function allows you to get the CloudBlue billing'
+            ' request data by the subscription ID or parameter value and sets'
+            ' item by ID or MPN.'
+        ),
+        edit_dialog_ui='/static/transformations/lookup_billing_request.html',
+    )
+    async def lookup_billing_request(
+        self,
+        row: Dict,
+    ):
+        output_columns = self.billing_settings.get('output_config')
+        value = row[self.billing_settings['parameter_column']]
+        item_id = row.get(self.billing_settings['item_column'])
+
+        if is_input_column_nullable(
+            self.transformation_request['transformation']['columns']['input'],
+            self.billing_settings['parameter_column'],
+        ) and not value:
+            return RowTransformationResponse.skip()
+
+        lookup = {}
+        if self.billing_settings.get('asset_type'):
+            lookup[f'asset.{self.billing_settings["asset_type"]}'] = row[
+                self.billing_settings['asset_column']
+            ]
+
+        lookup['asset.params.name'] = self.billing_settings['parameter']['name']
+        lookup['asset.params.value'] = value
+
+        try:
+            request = await self.get_billing_request(lookup)
+        except Exception as e:
+            return RowTransformationResponse.fail(output=str(e))
+
+        if not request:
+            return RowTransformationResponse.skip()
+
+        return RowTransformationResponse.done(
+            self.extract_row_from_billing(request, output_columns, item_id),
+        )
+
+    def extract_row_from_billing(self, request, output_columns, item_id):
+        row = {}
+
+        item_attrs = []
+
+        for col_name, col_config in output_columns.items():
+            value = None
+            attr = col_config['attribute']
+            if attr == 'asset.parameter.value':
+                param_name = col_config['parameter_name']
+                for param_data in request['asset']['params']:
+                    if param_data['name'] == param_name:
+                        value = param_data['value']
+                        break
+            elif attr.startswith('items.'):
+                item_attr = attr.split('.')[-1]
+                item_attrs.append((col_name, item_attr))
+                value = ''
+            else:
+                value = deep_itemgetter(request, attr)
+
+            row[col_name] = value
+
+        if item_attrs:
+            self.extract_item_attrs_from_billing(item_attrs, row, request, item_id)
+
+        return row
+
+    def extract_item_attrs_from_billing(self, item_attrs, row, request, item_id_value):
+        item_id = self.billing_settings['item']['id']
+        for item in request['items']:
+            if item_id == 'all' or item.get(item_id) == item_id_value:
+                for col_name, item_attr in item_attrs:
+                    item_value = str(item.get(item_attr, ''))
+                    row[col_name] += f'{SEPARATOR}{item_value}' if row[col_name] else item_value
+
+    async def get_billing_request(self, lookup):
+        k = 'billing-'
+        for key, value in lookup.items():
+            k = k + f'{key}-{value}'
+        try:
+            return self.cache_get(k)
+        except KeyError:
+            pass
+
+        result = await self.retrieve_billing_requests(lookup)
+
+        await self.acache_put(k, result)
+        return result
+
+    async def retrieve_billing_requests(self, lookup):
+        batch_context = self.transformation_request['batch']['context']
+        period_end = batch_context.get('period', {}).get('end')
+        additional_filters = {}
+        if period_end:
+            additional_filters['events.created.at__lt'] = period_end
+
+        requests = self.installation_client('subscriptions').requests.filter(
+            **lookup,
+            **additional_filters,
+        ).order_by('-events.created.at')
+        requests = [r async for r in requests]
+
+        result = None
+
+        for item in requests:
+            if result is None:
+                result = item
+            elif self.billing_settings.get('action_if_multiple') == 'leave_empty':
+                return
+            elif self.billing_settings.get('action_if_multiple') == 'use_most_actual':
+                return result
+            else:
+                raise BillingRequestLookupError(f'Many results found for the filter {lookup}')
+
+        if result:
+            return result
+
+        if self.billing_settings.get('action_if_not_found') == 'fail':
+            raise BillingRequestLookupError(f'No result found for the filter {lookup}')
+
+    @property
+    def billing_settings(self):
+        return self.transformation_request['transformation']['settings']
+
+
+class LookupBillingRequestWebAppMixin:
+
+    @router.post(
+        '/lookup_billing_request/validate',
+        summary='Validate lookup billing request settings',
+        response_model=ValidationResult,
+        responses={
+            400: {'model': Error},
+        },
+    )
+    def validate_lookup_billing_request_settings(
+        self,
+        data: Configuration,
+    ):
+        return validate_lookup_billing_request(data)
+
+    @router.get(
+        '/lookup_billing_request/parameters',
+        summary='Return available parameters names',
+    )
+    async def get_billing_parameters(
+        self,
+        product_id: str,
+        client: AsyncConnectClient = Depends(get_installation_client),
+    ):
+        return [
+            SubscriptionParameter(
+                id=parameter['id'],
+                name=parameter['name'],
+            )
+            async for parameter in client.products[product_id].parameters.all()
+        ]

connect_transformations/lookup_billing_request/mixins.py

@@ -0,0 +1,37 @@
+from typing import Dict, Optional
+
+from pydantic import BaseModel
+
+from connect_transformations.models import Columns
+
+
+class SubscriptionParameter(BaseModel):
+    id: str
+    name: str
+
+
+class Parameter(BaseModel):
+    id: Optional[str]
+    name: Optional[str]
+
+
+class OutputConfig(BaseModel):
+    attribute: Optional[str]
+    parameter: Optional[str]
+
+
+class Settings(BaseModel):
+    asset_type: Optional[str]
+    asset_column: Optional[str]
+    parameter: Optional[Parameter]
+    parameter_column: Optional[str]
+    item: Optional[Parameter]
+    item_column: Optional[str]
+    action_if_not_found: Optional[str]
+    action_if_multiple: Optional[str]
+    output_config: Optional[Dict[str, OutputConfig]]
+
+
+class Configuration(BaseModel):
+    settings: Optional[Settings]
+    columns: Optional[Columns]

connect_transformations/lookup_billing_request/models.py

@@ -0,0 +1,119 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2023, CloudBlue LLC
+# All rights reserved.
+#
+from connect_transformations.utils import (
+    build_error_response,
+    does_not_contain_required_keys,
+    has_invalid_basic_structure,
+)
+
+
+ASSET_LOOKUP = {
+    'external_id': 'CloudBlue Subscription External ID',
+    'id': 'CloudBlue Subscription ID',
+}
+
+
+NOT_FOUND_CHOICES = ['leave_empty', 'fail']
+MULTIPLE_CHOICES = ['leave_empty', 'fail', 'use_most_actual']
+
+
+def validate_lookup_billing_request(data):  # noqa: CCR001
+    data = data.dict(by_alias=True)
+
+    if has_invalid_basic_structure(data):
+        return build_error_response('Invalid input data')
+
+    if (
+        does_not_contain_required_keys(
+            data['settings'],
+            [
+                'parameter',
+                'parameter_column',
+                'action_if_not_found',
+                'action_if_multiple',
+                'output_config',
+            ],
+        )
+    ):
+        return build_error_response(
+            'The settings must have `parameter`, `parameter_column`, `item`, `item_column`, '
+            '`action_if_not_found` and `action_if_multiple` fields',
+        )
+
+    values = ASSET_LOOKUP.keys()
+    if data['settings'].get('asset_type') and data['settings'].get('asset_type') not in values:
+        return build_error_response(
+            f'The settings `asset_type` allowed values {", ".join(values)}',
+        )
+
+    if data['settings']['action_if_not_found'] not in NOT_FOUND_CHOICES:
+        return build_error_response(
+            f'The settings `action_if_not_found` allowed values {", ".join(NOT_FOUND_CHOICES)}',
+        )
+
+    if data['settings']['action_if_multiple'] not in MULTIPLE_CHOICES:
+        return build_error_response(
+            f'The settings `action_if_multiple` allowed values {", ".join(MULTIPLE_CHOICES)}',
+        )
+
+    if (
+        does_not_contain_required_keys(data['settings'], ['parameter'])
+        or not isinstance(data['settings']['parameter'], dict)
+        or does_not_contain_required_keys(data['settings']['parameter'], ['id', 'name'])
+    ):
+        return build_error_response('The settings must have `parameter` with `id` and `name`')
+
+    if (
+        does_not_contain_required_keys(data['settings'], ['item'])
+        or not isinstance(data['settings']['item'], dict)
+        or does_not_contain_required_keys(data['settings']['item'], ['id', 'name'])
+    ):
+        return build_error_response('The settings must have `item` with `id` and `name`')
+
+    input_columns = data['columns']['input']
+    available_input_columns = [c['name'] for c in input_columns]
+    columns_names = [data['settings']['parameter_column'], data['settings']['item_column']]
+    if data['settings'].get('asset_column'):
+        columns_names.append(data['settings']['asset_column'])
+    for column_name in columns_names:
+        if column_name not in available_input_columns:
+            return build_error_response(
+                f'The settings contains an invalid column name {column_name}'
+                ' that does not exist on columns.input',
+            )
+
+    output_columns = data['columns']['output']
+    output_columns_config = data['settings']['output_config']
+
+    for out_column in output_columns:
+        column_name = out_column['name']
+        column_config = output_columns_config.get(column_name)
+        if not column_config:
+            return build_error_response(
+                'The settings `output_config` does not contain '
+                f'settings for the column {column_name}.',
+            )
+
+    overview = 'Match by parameter "' + data['settings']['parameter']['name'] + '"\n'
+    if data['settings'].get('asset_type'):
+        overview += 'And "' + ASSET_LOOKUP[data['settings']['asset_type']] + '"\n'
+    overview += (
+        'If not found = '
+        + data['settings']['action_if_not_found'].replace('_', ' ').capitalize()
+        + '\n'
+    )
+    overview += (
+        'If multiple found = '
+        + data['settings']['action_if_multiple'].replace('_', ' ').capitalize()
+        + '\n'
+    )
+
+    total = len(data["columns"]["output"])
+    overview += f'And populate {total} column{"s" if total > 1 else ""}'
+
+    return {
+        'overview': overview,
+    }