Promotion Handler

Once the object is approved on the platform, the artifact-bundle gets generated. Corridor provides an option to promote the artifact bundle to any production environment, or send to any 3rd party application if required. The artifact information can even be processed, restructured, based on how the production environment or the 3rd party application requires the information. This can be accomplished by defining a custom handler which uses the base promotion handler class exposed by Corridor.

Handler class

The user would need to define a CustomPromotionHandler which would inherit Corridor's base promotion handler class: corridor_api.config.handlers.PromotionHandler

The logic for promotion would be defined the following method(s) inside CustomPromotionHandler.

• promote_bundle(io_bytes)

The name of the artifact bundle is also available with the PromotionHandler as attribute name.

Example

import datetime
import requests

from corridor_runtime.artifact import Artifact

from corridor_api.config.handlers import PromotionHandler


class CustomPromotionHandler(PromotionHandler):
    def promote_bundle(self, io_bytes):
        """
        :param io_bytes:  The artifact in io.BytesIO. This is a tar.gz file with the corridor artifact structure.
        """
        # Write the appropriate logic for production integration
        artifact_bundle_name = self.name
        artifact = Artifact(io_bytes)
        current_time = datetime.datetime.now()
        inputs = artifact.inputs
        input_info = {
            t.alias: {
                'id': t.id,
                'alias': t.alias,
                'name': t.name,
                'location': t.location,
                'columns': [col.alias for col in t.columns],
            }
            for t in artifact.inputs
        }
        info = {
            'artifact_name': artifact_bundle_name
            'inputs': input_info,
            'current_time': current_time,
            'artifact_code': artifact.code(lang_spec='pyspark-dataframe'),
        }
        headers = {}  # any headers can be configured (optional)
        # assuming /promote_artifact is the endpoint for the POST api in the 3rd party application running on port: 7006
        url = 'http://externaltool.example.com/promote_artifact'
        res = requests.post(url, files=info, headers=headers)

        if res.status_code != 200:
            raise ValueError('Artifact promotion failed!')

Configurations

Promotion handler related configurations need to be set in api_config.py along with other configurations (assuming the CustomPromotionHandler class is defined in the file custom_promotion_handler.py).

PROMOTION_HANDLER = 'custom_promotion_handler.CustomPromotionHandler'