tessl/pypi-google-cloud-retail
Python client library for Google Cloud Retail API enabling end-to-end personalized recommendation systems
—
Pending
product-management.mddocs/
Product Management
Comprehensive product catalog operations including creating, updating, and managing product information, inventory, and fulfillment details. The Product Service provides both synchronous and asynchronous clients for optimal performance in different application contexts.
Capabilities
Basic Product Operations
Core CRUD operations for managing products in your retail catalog.
class ProductServiceClient:
def create_product(self, request: CreateProductRequest) -> Product:
"""
Creates a product in the catalog.
Args:
request: Contains parent (catalog/branch path), product data, and product_id
Returns:
Product: The created product with generated name and metadata
Raises:
AlreadyExists: If product_id already exists
InvalidArgument: If required fields are missing or invalid
"""
def get_product(self, request: GetProductRequest) -> Product:
"""
Retrieves a product by its resource name.
Args:
request: Contains name (full product resource path)
Returns:
Product: The requested product
Raises:
NotFound: If product doesn't exist
"""
def list_products(self, request: ListProductsRequest) -> ListProductsResponse:
"""
Lists products in a catalog branch with optional filtering.
Args:
request: Contains parent path, page_size, page_token, filter, read_mask
Returns:
ListProductsResponse: Products and next_page_token for pagination
"""
def update_product(self, request: UpdateProductRequest) -> Product:
"""
Updates an existing product.
Args:
request: Contains product data and update_mask for partial updates
Returns:
Product: The updated product
Raises:
NotFound: If product doesn't exist
"""
def delete_product(self, request: DeleteProductRequest) -> None:
"""
Deletes a product from the catalog.
Args:
request: Contains name (product resource path)
Raises:
NotFound: If product doesn't exist
"""
class ProductServiceAsyncClient:
async def create_product(self, request: CreateProductRequest) -> Product:
"""Async version of create_product."""
async def get_product(self, request: GetProductRequest) -> Product:
"""Async version of get_product."""
async def list_products(self, request: ListProductsRequest) -> ListProductsResponse:
"""Async version of list_products."""
async def update_product(self, request: UpdateProductRequest) -> Product:
"""Async version of update_product."""
async def delete_product(self, request: DeleteProductRequest) -> None:
"""Async version of delete_product."""Inventory Management
Operations for managing product inventory, availability, and fulfillment locations.
class ProductServiceClient:
def set_inventory(self, request: SetInventoryRequest) -> Operation:
"""
Updates product inventory information (long-running operation).
Args:
request: Contains inventory data, set_mask, and optional set_time
Returns:
Operation: Long-running operation that resolves to SetInventoryResponse
"""
def add_fulfillment_places(self, request: AddFulfillmentPlacesRequest) -> Operation:
"""
Adds fulfillment places to a product (long-running operation).
Args:
request: Contains product name, type, place_ids, and add_time
Returns:
Operation: Resolves to AddFulfillmentPlacesResponse
"""
def remove_fulfillment_places(self, request: RemoveFulfillmentPlacesRequest) -> Operation:
"""
Removes fulfillment places from a product (long-running operation).
Args:
request: Contains product name, type, place_ids, and remove_time
Returns:
Operation: Resolves to RemoveFulfillmentPlacesResponse
"""
def add_local_inventories(self, request: AddLocalInventoriesRequest) -> Operation:
"""
Adds local inventory information to a product (long-running operation).
Args:
request: Contains product name, local_inventories, and add_time
Returns:
Operation: Resolves to AddLocalInventoriesResponse
"""
def remove_local_inventories(self, request: RemoveLocalInventoriesRequest) -> Operation:
"""
Removes local inventory information from a product (long-running operation).
Args:
request: Contains product name, place_ids, and remove_time
Returns:
Operation: Resolves to RemoveLocalInventoriesResponse
"""
class ProductServiceAsyncClient:
async def set_inventory(self, request: SetInventoryRequest) -> Operation:
"""Async version of set_inventory."""
async def add_fulfillment_places(self, request: AddFulfillmentPlacesRequest) -> Operation:
"""Async version of add_fulfillment_places."""
async def remove_fulfillment_places(self, request: RemoveFulfillmentPlacesRequest) -> Operation:
"""Async version of remove_fulfillment_places."""
async def add_local_inventories(self, request: AddLocalInventoriesRequest) -> Operation:
"""Async version of add_local_inventories."""
async def remove_local_inventories(self, request: RemoveLocalInventoriesRequest) -> Operation:
"""Async version of remove_local_inventories."""Bulk Operations
Large-scale operations for importing and purging products in batches.
class ProductServiceClient:
def import_products(self, request: ImportProductsRequest) -> Operation:
"""
Imports products in bulk from external sources (long-running operation).
Args:
request: Contains parent, input_config, errors_config, and update_mask
Returns:
Operation: Resolves to ImportProductsResponse with import statistics
"""
def purge_products(self, request: PurgeProductsRequest) -> Operation:
"""
Permanently deletes products matching filter criteria (long-running operation).
Args:
request: Contains parent, filter, and force flag
Returns:
Operation: Resolves to PurgeProductsResponse with purge count
"""
class ProductServiceAsyncClient:
async def import_products(self, request: ImportProductsRequest) -> Operation:
"""Async version of import_products."""
async def purge_products(self, request: PurgeProductsRequest) -> Operation:
"""Async version of purge_products."""Data Types
Product
Core product entity containing all product information and metadata.
class Product:
name: str # Resource name (read-only)
id: str # Product ID (required)
type: ProductType # PRIMARY or VARIANT
primary_product_id: str # For variant products
collection_member_ids: List[str] # Related products
gtin: str # Global Trade Item Number
categories: List[str] # Product categories
title: str # Product title (required)
brands: List[str] # Brand names
description: str # Product description
language_code: str # Content language
attributes: Dict[str, CustomAttribute] # Custom attributes
tags: List[str] # Tags for organization
price_info: PriceInfo # Pricing information
rating: Rating # Average rating and review count
available_time: Timestamp # When product becomes available
availability: ProductAvailability # IN_STOCK, OUT_OF_STOCK, etc.
available_quantity: int # Quantity available
fulfillment_info: List[FulfillmentInfo] # Fulfillment options
uri: str # Product page URL
images: List[Image] # Product images
audience: Audience # Target audience
color_info: ColorInfo # Color information
sizes: List[str] # Available sizes
materials: List[str] # Materials used
patterns: List[str] # Patterns/designs
conditions: List[str] # Product conditions
promotions: List[Promotion] # Active promotions
publish_time: Timestamp # Publication time
retrievable_fields: FieldMask # Fields included in search
variants: List[Product] # Product variants (read-only)
local_inventories: List[LocalInventory] # Local inventory dataSupporting Types
class PriceInfo:
currency_code: str # ISO 4217 currency code (required)
price: float # Listed price
original_price: float # Original price before discounts
cost: float # Cost price
price_effective_time: Timestamp # When price becomes effective
price_expire_time: Timestamp # When price expires
price_range: PriceInfoPriceRange # Price range for variants
class CustomAttribute:
text: List[str] # Text values
numbers: List[float] # Numeric values
searchable: bool # Whether attribute is searchable
indexable: bool # Whether attribute is indexable
class FulfillmentInfo:
type: str # Fulfillment type (pickup-in-store, ship-to-store, etc.)
place_ids: List[str] # Location identifiers
class LocalInventory:
place_id: str # Store/location identifier
price_info: PriceInfo # Local pricing
attributes: Dict[str, CustomAttribute] # Local attributes
fulfillment_types: List[str] # Available fulfillment types
class Rating:
rating_count: int # Total number of ratings
average_rating: float # Average rating value
rating_histogram: List[int] # Rating distribution (1-5 stars)
class Image:
uri: str # Image URL
height: int # Image height in pixels
width: int # Image width in pixelsRequest Types
class CreateProductRequest:
parent: str # Catalog branch path (required)
product: Product # Product to create (required)
product_id: str # Product ID (required)
class GetProductRequest:
name: str # Product resource name (required)
class ListProductsRequest:
parent: str # Catalog branch path (required)
page_size: int # Maximum products to return (default: 100, max: 1000)
page_token: str # Token for pagination
filter: str # Filter expression
read_mask: FieldMask # Fields to return
class UpdateProductRequest:
product: Product # Product with updates (required)
update_mask: FieldMask # Fields to update
allow_missing: bool # Create if product doesn't exist
class DeleteProductRequest:
name: str # Product resource name (required)
class SetInventoryRequest:
inventory: Product # Product with inventory updates (required)
set_mask: FieldMask # Inventory fields to update
set_time: Timestamp # When inventory change occurred
allow_missing: bool # Create product if doesn't exist
class ImportProductsRequest:
parent: str # Catalog branch path (required)
request_id: str # Unique request identifier
input_config: ProductInputConfig # Data source configuration
errors_config: ImportErrorsConfig # Error handling configuration
update_mask: FieldMask # Fields to update for existing products
reconciliation_mode: ImportProductsRequestReconciliationMode # How to handle existing products
notification_pubsub_topic: str # Pub/Sub topic for completion notificationUsage Examples
Creating a Product
from google.cloud import retail
client = retail.ProductServiceClient()
# Create a product with comprehensive information
product = retail.Product(
id="electronics-laptop-123",
type=retail.Product.Type.PRIMARY,
title="Gaming Laptop Pro 15",
categories=["Electronics", "Computers", "Laptops"],
brands=["TechBrand"],
description="High-performance gaming laptop with advanced graphics",
price_info=retail.PriceInfo(
currency_code="USD",
price=1299.99,
original_price=1499.99
),
availability=retail.Product.Availability.IN_STOCK,
available_quantity=25,
attributes={
"screen_size": retail.CustomAttribute(text=["15.6 inches"]),
"processor": retail.CustomAttribute(text=["Intel i7"]),
"ram": retail.CustomAttribute(text=["16GB"]),
"storage": retail.CustomAttribute(text=["512GB SSD"]),
"color": retail.CustomAttribute(text=["Black"])
},
tags=["gaming", "high-performance", "laptop"],
uri="https://example.com/products/gaming-laptop-pro-15"
)
request = retail.CreateProductRequest(
parent="projects/my-project/locations/global/catalogs/default_catalog/branches/default_branch",
product=product,
product_id=product.id
)
created_product = client.create_product(request=request)
print(f"Created product: {created_product.name}")Updating Product Inventory
from google.protobuf import field_mask_pb2
# Update inventory for a product
inventory_update = retail.Product(
name="projects/my-project/locations/global/catalogs/default_catalog/branches/default_branch/products/electronics-laptop-123",
availability=retail.Product.Availability.OUT_OF_STOCK,
available_quantity=0
)
request = retail.SetInventoryRequest(
inventory=inventory_update,
set_mask=field_mask_pb2.FieldMask(paths=["availability", "available_quantity"])
)
operation = client.set_inventory(request=request)
print(f"Inventory update operation: {operation.name}")
# Wait for operation to complete
result = operation.result()
print(f"Inventory updated: {result}")Bulk Product Import
# Import products from Google Cloud Storage
input_config = retail.ProductInputConfig(
gcs_source=retail.GcsSource(
input_uris=["gs://my-bucket/products/products.json"]
)
)
errors_config = retail.ImportErrorsConfig(
gcs_prefix="gs://my-bucket/errors/"
)
request = retail.ImportProductsRequest(
parent="projects/my-project/locations/global/catalogs/default_catalog/branches/default_branch",
input_config=input_config,
errors_config=errors_config,
reconciliation_mode=retail.ImportProductsRequest.ReconciliationMode.INCREMENTAL
)
operation = client.import_products(request=request)
print(f"Import operation: {operation.name}")
# Monitor operation progress
result = operation.result()
print(f"Import completed. Success count: {result.success_count}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-retail