def process()

in python/activation/main.py [0:0]

• 26 lines of code
• 3 McCabe index (conditional complexity)

  def process(self, element):
    """
    Transforms the output of the Measurement Protocol API call into a format suitable for logging.

    Args:
      element: A tuple containing the event that was sent and the HTTP status code of the response.

    Yields:
      A dictionary containing the following fields:
        - id: A unique identifier for the log entry.
        - activation_id: The ID of the activation event.
        - payload: The JSON payload of the event that was sent.
        - latest_state: The latest state of the event, which can be either "SEND_OK" or "SEND_FAIL".
        - updated_at: The timestamp when the log entry was created.
    """
    time_cast = datetime.datetime.now(tz=datetime.timezone.utc)

    if element[1] == requests.status_codes.codes.NO_CONTENT:
      state_msg = 'SEND_OK'
    else:
      state_msg = 'SEND_FAIL'

    result = {}
    try:
      result = {
        'id': str(uuid.uuid4()),
        'activation_id': element[0]['events'][0]['name'],
        'payload': json.dumps(element[0]),
        'latest_state': f"{state_msg} {element[1]}",
        'updated_at': str(time_cast)
      }
    except KeyError as e:
      logging.error(element)
      result = {
        'id': str(uuid.uuid4()),
        'activation_id': "",
        'payload': json.dumps(element[0]),
        'latest_state': f"{state_msg} {element[1]}",
        'updated_at': str(time_cast)
      }
      logging.error(traceback.format_exc())
    yield result