Repository: microsoft/promptflow
File: src/promptflow-azure/promptflow/azure/_storage/cosmosdb/span.py (Python)

Code before:
            return cosmos_client.create_item(body=self.to_cosmosdb_item())
        except CosmosResourceExistsError:
            return

    def to_dict(self) -> Dict[str, Any]:
        return {k: v for k, v in self.__dict__.items() if v}

    def to_cosmosdb_item(self, attr_value_truncation_length: int = 8 * 1024):
        """
        Convert the object to a dictionary for persistence to CosmosDB.
        Truncate attribute values to avoid exceeding CosmosDB's 2MB size limit.
        """
        item = self.to_dict()  # use to_dict method to get a dictionary representation of the object

        # Serialize the dictionary to a JSON string using separators (",", ":").
        # This is done to align with the serialization method used by the CosmosDB Python SDK.
        json_dumped = json.dumps(item, separators=(",", ":"))

        item_size = len(json_dumped.encode("utf-8"))
        max_size_in_bytes = 2 * 1024 * 1024  # 2MB in bytes

        if item_size > max_size_in_bytes:
            attributes = item.get("attributes")
            if attributes:
                item["attributes"] = {
                    k: (v if not isinstance(v, str) else v[:attr_value_truncation_length])
                    for k, v in attributes.items()
                }
        return item

    def _persist_events(self, blob_container_client: ContainerClient, blob_base_uri: str):
        for idx, event in enumerate(self.events):
            event_data = json.dumps(event)
            blob_client = blob_container_client.get_blob_client(self._event_path(idx))
            blob_client.upload_blob(event_data)

            event[SpanEventFieldName.ATTRIBUTES] = {}
            self.external_event_data_uris.append(f"{blob_base_uri}{self._event_path(idx)}")

    EVENT_PATH_PREFIX = ".promptflow/.trace"

    def _event_path(self, idx: int) -> str:
        trace_id = self.context[SpanContextFieldName.TRACE_ID]
        return f"{self.EVENT_PATH_PREFIX}/{self.collection_id}/{trace_id}/{self.id}/{idx}"


Review comment (suggestion):
How about
```suggestion
                    k: (v[:attr_value_truncation_length] if isinstance(v, str) else v)
```

Code after:
            return cosmos_client.create_item(body=self.to_cosmosdb_item())
        except CosmosResourceExistsError:
            return

    def to_dict(self) -> Dict[str, Any]:
        return {k: v for k, v in self.__dict__.items() if v}

    def to_cosmosdb_item(self, attr_value_truncation_length: int = 8 * 1024):
        """
        Convert the object to a dictionary for persistence to CosmosDB.
        Truncate attribute values to avoid exceeding CosmosDB's 2MB size limit.
        """
        item = self.to_dict()  # use to_dict method to get a dictionary representation of the object

        attributes = item.get("attributes")
        if attributes:
            item["attributes"] = {
                k: (v[:attr_value_truncation_length] if isinstance(v, str) else v) for k, v in attributes.items()
            }
        return item

    def _persist_events(self, blob_container_client: ContainerClient, blob_base_uri: str):
        for idx, event in enumerate(self.events):
            event_data = json.dumps(event)
            blob_client = blob_container_client.get_blob_client(self._event_path(idx))
            blob_client.upload_blob(event_data)

            event[SpanEventFieldName.ATTRIBUTES] = {}
            self.external_event_data_uris.append(f"{blob_base_uri}{self._event_path(idx)}")

    EVENT_PATH_PREFIX = ".promptflow/.trace"

    def _event_path(self, idx: int) -> str:
        trace_id = self.context[SpanContextFieldName.TRACE_ID]
        return f"{self.EVENT_PATH_PREFIX}/{self.collection_id}/{trace_id}/{self.id}/{idx}"