def store_product()

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

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

    def store_product(self, product_id: str, product_data: Dict[str, Any]) -> bool:
        """
        Store product data in Spanner

        Args:
            product_id: Unique product identifier
            product_data: Product metadata

        Returns:
            bool: True if successful, False otherwise
        """
        try:
            # Convert non-serializable objects to serializable format
            product_json = self._prepare_product_for_spanner(product_data)

            # Prepare mutation to insert or update product
            with self.database.batch() as batch:
                batch.insert_or_update(
                    table="products",
                    columns=[
                        "product_id",
                        "product_data",
                        "title",
                        "description",
                        "timestamp",
                    ],
                    values=[
                        (
                            product_id,
                            product_json,
                            product_data.get("title", ""),
                            product_data.get("description", ""),
                            spanner.COMMIT_TIMESTAMP,
                        )
                    ],
                )

            logging.info(f"Successfully stored product {product_id} in Spanner")
            return True

        except Exception as e:
            logging.error(f"Error storing product {product_id} in Spanner: {str(e)}")
            return False