def write()

in src/functions_framework/execution_id.py [0:0]

• 28 lines of code
• 9 McCabe index (conditional complexity)

    def write(self, contents):
        if contents == "\n":
            return
        current_context = _get_current_context()
        if current_context is None:
            self.stream.write(contents + "\n")
            self.stream.flush()
            return
        try:
            execution_id = current_context.execution_id
            span_id = current_context.span_id
            payload = json.loads(contents)
            if not isinstance(payload, dict):
                payload = {"message": contents}
        except json.JSONDecodeError:
            if len(contents) > 0 and contents[-1] == "\n":
                contents = contents[:-1]
            payload = {"message": contents}
        if execution_id:
            payload[_LOGGING_API_LABELS_FIELD] = payload.get(
                _LOGGING_API_LABELS_FIELD, {}
            )
            payload[_LOGGING_API_LABELS_FIELD]["execution_id"] = execution_id
        if span_id:
            payload[_LOGGING_API_SPAN_ID_FIELD] = span_id
        self.stream.write(json.dumps(payload))
        self.stream.write("\n")
        self.stream.flush()