def conversion_actions()

in athena_glue_service_logs/cloudtrail.py [0:0]

• 10 lines of code
• 5 McCabe index (conditional complexity)

    def conversion_actions(self, dynamic_frame):
        LOGGER.info("Performing custom conversion action: to_json")

        # The list of fields and mapping tuples we want to convert to a raw JSON string
        raw_fields = ['requestParameters', 'responseElements', 'userIdentity', 'additionalEventData']
        json_mappings = [(name, 'struct', "json_%s" % name.lower(), 'string') for name in raw_fields]

        # Keep the remaining fields the exact same
        static_mappings = [
            (f.name, f.dataType.typeName(), f.name, f.dataType.typeName())
            for f in dynamic_frame.schema() if f.name not in raw_fields
        ]

        # Apply the mapping of the combined field set
        mapped_dyf = dynamic_frame.apply_mapping(static_mappings + json_mappings)

        return mapped_dyf