def _prepare_product_for_spanner()

in src/psearch/ingestion/services/spanner_service.py [0:0]

• 34 lines of code
• 18 McCabe index (conditional complexity)

    def _prepare_product_for_spanner(self, product_data: Dict[str, Any]) -> str:
        """
        Prepares product data for storage in Spanner by converting to JSON
        and handling non-serializable values

        Args:
            product_data: Raw product data

        Returns:
            str: JSON string representation of product data
        """
        # Create a copy to avoid modifying the original
        product_copy = product_data.copy()

        # Recursively sanitize the product data
        def sanitize_value(value):
            if isinstance(value, dict):
                return {k: sanitize_value(v) for k, v in value.items()}
            elif isinstance(value, list):
                return [sanitize_value(item) for item in value]
            elif isinstance(value, (int, bool, str)) or value is None:
                return value
            elif isinstance(value, float):
                # Handle NaN and Infinity which Spanner doesn't accept
                if math.isnan(value) or math.isinf(value):
                    return None

                # Handle very large or very small numbers that might cause issues
                if abs(value) > 1e15:  # Very large numbers
                    # Round to avoid precision issues
                    return round(value, 2)
                elif 0 < abs(value) < 1e-10:  # Very small non-zero numbers
                    # Round to avoid precision issues with very small numbers
                    return 0.0

                # For price-like values (common in product data), round to 2 decimal places
                if 0.01 <= abs(value) <= 10000.0:
                    return round(value, 2)

                # For other normal range numbers, ensure we don't have too many decimal places
                return round(value, 6)
            else:
                # Convert any other type to string
                return str(value)

        # Sanitize the entire product
        sanitized_product = sanitize_value(product_copy)

        # Special handling for priceInfo to ensure consistent formatting
        if "priceInfo" in sanitized_product:
            price_info = sanitized_product["priceInfo"]
            if "price" in price_info and isinstance(price_info["price"], float):
                price_info["price"] = round(price_info["price"], 2)
            if "cost" in price_info and isinstance(price_info["cost"], float):
                price_info["cost"] = round(price_info["cost"], 2)
            if "originalPrice" in price_info and isinstance(
                price_info["originalPrice"], float
            ):
                price_info["originalPrice"] = round(price_info["originalPrice"], 2)

        # Convert to JSON string with special handling for Spanner
        json_str = json.dumps(sanitized_product, ensure_ascii=True)

        return json_str