def create_event()

in CommonLayerCode/datalake-library/python/datalake_library/octagon/event.py [0:0]

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

    def create_event(self, reason, comment, component_name=None, event_details=None):

        throw_if_false(self.client.is_pipeline_set(), "Pipeline execution is not yet assigned")

        throw_none_or_empty(str(reason), "No reason specified")
        throw_none_or_empty(comment, "No comment specified")

        self.logger.debug(f"event create_event() called. Reason : {reason}, comment: {comment}")

        if isinstance(reason, EventReasonEnum):
            reason_str = reason.value
        else:
            reason_str = reason

        # Put to DDB

        item = {}
        current_time = datetime.datetime.utcnow()
        local_date_iso = get_local_date()
        utc_time_iso = get_timestamp_iso(current_time)
        # Add extra fields
        item["id"] = str(uuid.uuid4())
        item["pipeline_execution_id"] = self.client.pipeline_execution_id
        item["pipeline"] = self.client.pipeline_name
        item["reason"] = reason_str
        item["comment"] = comment
        item["timestamp"] = utc_time_iso
        item["date"] = local_date_iso
        item["date_and_reason"] = item["date"] + "#" + item["reason"]
        if component_name:
            item["component"] = component_name
        if event_details:
            item["details"] = event_details

        if self.events_ttl != 0:
            item["ttl"] = get_ttl(self.events_ttl)
        # Save to DDB
        self.events_table.put_item(Item=item)
        return item["id"]