Product Search and Catalog Management

class ProductSearchClient:
    def __init__(self, *, credentials=None, transport=None, client_options=None, client_info=None):
        """Initialize the ProductSearchClient.
        
        Args:
            credentials: The authorization credentials to attach to requests
            transport: The transport to use for API calls
            client_options: Custom options for the client
            client_info: The client info used to send a user-agent string
        """

from google.cloud.vision import ProductSearchClient, Product, ProductSet, ReferenceImage

# Initialize client
client = ProductSearchClient()
location_path = f"projects/{project_id}/locations/{location_id}"

# Create a product set
product_set = ProductSet(display_name="Clothing", description="Fashion items")
response = client.create_product_set(
    parent=location_path,
    product_set=product_set,
    product_set_id="clothing_set_1"
)

# Create a product
product = Product(
    display_name="Blue Jeans",
    description="Classic blue denim jeans",
    product_category="apparel"
)
product_response = client.create_product(
    parent=location_path,
    product=product,
    product_id="blue_jeans_1"  
)

# Add product to product set
client.add_product_to_product_set(
    name=response.name,
    product=product_response.name
)

def create_product_set(self, parent, product_set, product_set_id=None, *, retry=None, timeout=None, metadata=()) -> ProductSet:
    """Create a new ProductSet.
    
    Args:
        parent: Project and location path (projects/{project}/locations/{location})
        product_set: ProductSet object with display_name and optional description
        product_set_id: User-defined ID for the product set
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ProductSet: The created product set with assigned name
    """

def list_product_sets(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductSetsResponse:
    """List ProductSets in a location.
    
    Args:
        parent: Project and location path
        page_size: Maximum number of results per page
        page_token: Token for pagination
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ListProductSetsResponse: Paginated list of product sets
    """

def get_product_set(self, name, *, retry=None, timeout=None, metadata=()) -> ProductSet:
    """Get information about a ProductSet.
    
    Args:
        name: Full resource name of the product set
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ProductSet: The requested product set
    """

def update_product_set(self, product_set, *, update_mask=None, retry=None, timeout=None, metadata=()) -> ProductSet:
    """Update a ProductSet.
    
    Args:
        product_set: ProductSet object with updates
        update_mask: FieldMask indicating which fields to update
        retry: Retry configuration for the request  
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ProductSet: The updated product set
    """

def delete_product_set(self, name, *, retry=None, timeout=None, metadata=()) -> None:
    """Delete a ProductSet.
    
    Args:
        name: Full resource name of the product set
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        None
    """

def create_product(self, parent, product, product_id=None, *, retry=None, timeout=None, metadata=()) -> Product:
    """Create a new Product.
    
    Args:
        parent: Project and location path
        product: Product object with display_name, description, product_category
        product_id: User-defined ID for the product
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        Product: The created product with assigned name
    """

def list_products(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductsResponse:
    """List Products in a location.
    
    Args:
        parent: Project and location path
        page_size: Maximum number of results per page
        page_token: Token for pagination
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ListProductsResponse: Paginated list of products
    """

def get_product(self, name, *, retry=None, timeout=None, metadata=()) -> Product:
    """Get information about a Product.
    
    Args:
        name: Full resource name of the product
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        Product: The requested product
    """

def update_product(self, product, *, update_mask=None, retry=None, timeout=None, metadata=()) -> Product:
    """Update a Product.
    
    Args:
        product: Product object with updates
        update_mask: FieldMask indicating which fields to update
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        Product: The updated product
    """

def delete_product(self, name, *, retry=None, timeout=None, metadata=()) -> None:
    """Delete a Product.
    
    Args:
        name: Full resource name of the product
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        None
    """

def create_reference_image(self, parent, reference_image, reference_image_id=None, *, retry=None, timeout=None, metadata=()) -> ReferenceImage:
    """Create a new ReferenceImage for a Product.
    
    Args:
        parent: Product path (projects/{project}/locations/{location}/products/{product})
        reference_image: ReferenceImage object with URI
        reference_image_id: User-defined ID for the reference image
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ReferenceImage: The created reference image
    """

def list_reference_images(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListReferenceImagesResponse:
    """List ReferenceImages for a Product.
    
    Args:
        parent: Product path
        page_size: Maximum number of results per page
        page_token: Token for pagination
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ListReferenceImagesResponse: Paginated list of reference images
    """

def get_reference_image(self, name, *, retry=None, timeout=None, metadata=()) -> ReferenceImage:
    """Get information about a ReferenceImage.
    
    Args:
        name: Full resource name of the reference image
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ReferenceImage: The requested reference image
    """

def delete_reference_image(self, name, *, retry=None, timeout=None, metadata=()) -> None:
    """Delete a ReferenceImage.
    
    Args:
        name: Full resource name of the reference image
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        None
    """

def add_product_to_product_set(self, name, product, *, retry=None, timeout=None, metadata=()) -> None:
    """Add a Product to a ProductSet.
    
    Args:
        name: ProductSet resource name
        product: Product resource name to add
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        None
    """

def remove_product_from_product_set(self, name, product, *, retry=None, timeout=None, metadata=()) -> None:
    """Remove a Product from a ProductSet.
    
    Args:
        name: ProductSet resource name
        product: Product resource name to remove
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        None
    """

def list_products_in_product_set(self, name, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductsInProductSetResponse:
    """List Products in a ProductSet.
    
    Args:
        name: ProductSet resource name
        page_size: Maximum number of results per page
        page_token: Token for pagination
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        ListProductsInProductSetResponse: Paginated list of products in the set
    """

def import_product_sets(self, parent, input_config, *, retry=None, timeout=None, metadata=()) -> Operation:
    """Import ProductSets from a CSV file.
    
    Args:
        parent: Project and location path
        input_config: ImportProductSetsInputConfig with CSV source details
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        Operation: Long-running operation for the import job
    """

def purge_products(self, parent, *, product_set_purge_config=None, delete_orphan_products=None, force=None, retry=None, timeout=None, metadata=()) -> Operation:
    """Delete products and their reference images.
    
    Args:
        parent: Project and location path
        product_set_purge_config: Configuration for purging products in a set
        delete_orphan_products: Whether to delete orphaned products
        force: Whether to force deletion
        retry: Retry configuration for the request
        timeout: Timeout for the request in seconds
        metadata: Additional metadata to send with the request
        
    Returns:
        Operation: Long-running operation for the purge job
    """

from google.cloud.vision import ProductSearchClient, Product, ProductSet, ReferenceImage

client = ProductSearchClient()
project_id = "your-project-id"
location_id = "us-west1"
location_path = f"projects/{project_id}/locations/{location_id}"

# Create product set for clothing items
product_set = ProductSet(
    display_name="Fashion Catalog",
    description="Collection of clothing and accessories"
)

created_set = client.create_product_set(
    parent=location_path,
    product_set=product_set,
    product_set_id="fashion_catalog_v1"
)

print(f"Created product set: {created_set.name}")

# Create products
products = [
    {
        "id": "shirt_001", 
        "display_name": "Blue Cotton Shirt",
        "description": "Classic blue cotton button-down shirt",
        "category": "apparel"
    },
    {
        "id": "jeans_001",
        "display_name": "Dark Wash Jeans", 
        "description": "Slim fit dark wash denim jeans",
        "category": "apparel"
    }
]

created_products = []
for prod_info in products:
    product = Product(
        display_name=prod_info["display_name"],
        description=prod_info["description"],
        product_category=prod_info["category"]
    )
    
    created_product = client.create_product(
        parent=location_path,
        product=product,
        product_id=prod_info["id"]
    )
    created_products.append(created_product)
    
    # Add to product set
    client.add_product_to_product_set(
        name=created_set.name,
        product=created_product.name
    )
    
    print(f"Created product: {created_product.display_name}")

# Add reference images for each product
reference_images = [
    {
        "product_name": created_products[0].name,
        "image_uri": "gs://your-bucket/shirt_front.jpg",
        "ref_id": "shirt_001_front"
    },
    {
        "product_name": created_products[0].name, 
        "image_uri": "gs://your-bucket/shirt_back.jpg",
        "ref_id": "shirt_001_back"
    },
    {
        "product_name": created_products[1].name,
        "image_uri": "gs://your-bucket/jeans_front.jpg", 
        "ref_id": "jeans_001_front"
    }
]

for ref_info in reference_images:
    reference_image = ReferenceImage(uri=ref_info["image_uri"])
    
    created_ref = client.create_reference_image(
        parent=ref_info["product_name"],
        reference_image=reference_image,
        reference_image_id=ref_info["ref_id"]
    )
    
    print(f"Added reference image: {created_ref.name}")

from google.cloud.vision import ImageAnnotatorClient, Image, ImageContext, ProductSearchParams

# Use ImageAnnotatorClient for product search
image_client = ImageAnnotatorClient()

# Configure product search parameters
product_search_params = ProductSearchParams(
    product_set=created_set.name,
    product_categories=["apparel"]
)

image_context = ImageContext(product_search_params=product_search_params)

# Search for products in an image
image = Image(source={'image_uri': 'gs://your-bucket/query_image.jpg'})

response = image_client.product_search(
    image=image,
    image_context=image_context,
    max_results=10
)

# Process search results  
if response.product_search_results:
    results = response.product_search_results
    print(f"Found {len(results.results)} matching products")
    
    for result in results.results:
        product = result.product
        print(f"Product: {product.display_name}")
        print(f"Score: {result.score}")
        print(f"Image: {result.image}")

# List all product sets
response = client.list_product_sets(parent=location_path)
for product_set in response.product_sets:
    print(f"Product Set: {product_set.display_name}")
    
    # List products in each set
    products_response = client.list_products_in_product_set(
        name=product_set.name
    )
    
    for product in products_response.products:
        print(f"  Product: {product.display_name}")
        
        # List reference images for each product  
        images_response = client.list_reference_images(
            parent=product.name
        )
        
        for ref_image in images_response.reference_images:
            print(f"    Reference: {ref_image.uri}")

from google.cloud.vision import ImportProductSetsInputConfig, ImportProductSetsGcsSource

# Import products from CSV file
gcs_source = ImportProductSetsGcsSource(
    csv_file_uri="gs://your-bucket/products.csv"
)

input_config = ImportProductSetsInputConfig(gcs_source=gcs_source)

operation = client.import_product_sets(
    parent=location_path,
    input_config=input_config
)

print(f"Import operation: {operation.name}")

# Wait for operation to complete
result = operation.result()
print(f"Import completed: {result}")

# Remove product from product set
client.remove_product_from_product_set(
    name=created_set.name,
    product=created_products[0].name
)

# Delete reference images
for product in created_products:
    ref_images = client.list_reference_images(parent=product.name)
    for ref_image in ref_images.reference_images:
        client.delete_reference_image(name=ref_image.name)

# Delete products
for product in created_products:
    client.delete_product(name=product.name)

# Delete product set
client.delete_product_set(name=created_set.name)

print("Cleanup completed")

class ProductSearchAsyncClient:
    async def create_product_set(self, parent, product_set, product_set_id=None, *, retry=None, timeout=None, metadata=()) -> ProductSet: ...
    async def list_product_sets(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductSetsResponse: ...
    async def get_product_set(self, name, *, retry=None, timeout=None, metadata=()) -> ProductSet: ...
    async def update_product_set(self, product_set, *, update_mask=None, retry=None, timeout=None, metadata=()) -> ProductSet: ...
    async def delete_product_set(self, name, *, retry=None, timeout=None, metadata=()) -> None: ...
    
    async def create_product(self, parent, product, product_id=None, *, retry=None, timeout=None, metadata=()) -> Product: ...
    async def list_products(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductsResponse: ...
    async def get_product(self, name, *, retry=None, timeout=None, metadata=()) -> Product: ...
    async def update_product(self, product, *, update_mask=None, retry=None, timeout=None, metadata=()) -> Product: ...
    async def delete_product(self, name, *, retry=None, timeout=None, metadata=()) -> None: ...
    
    async def create_reference_image(self, parent, reference_image, reference_image_id=None, *, retry=None, timeout=None, metadata=()) -> ReferenceImage: ...
    async def list_reference_images(self, parent, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListReferenceImagesResponse: ...
    async def get_reference_image(self, name, *, retry=None, timeout=None, metadata=()) -> ReferenceImage: ...
    async def delete_reference_image(self, name, *, retry=None, timeout=None, metadata=()) -> None: ...
    
    async def add_product_to_product_set(self, name, product, *, retry=None, timeout=None, metadata=()) -> None: ...
    async def remove_product_from_product_set(self, name, product, *, retry=None, timeout=None, metadata=()) -> None: ...
    async def list_products_in_product_set(self, name, *, page_size=None, page_token=None, retry=None, timeout=None, metadata=()) -> ListProductsInProductSetResponse: ...
    async def import_product_sets(self, parent, input_config, *, retry=None, timeout=None, metadata=()) -> Operation: ...
    async def purge_products(self, parent, *, product_set_purge_config=None, delete_orphan_products=None, force=None, retry=None, timeout=None, metadata=()) -> Operation: ...

import asyncio
from google.cloud.vision import ProductSearchAsyncClient, Product

async def manage_products_async():
    client = ProductSearchAsyncClient()
    location_path = "projects/your-project/locations/us-west1"
    
    # Create product asynchronously
    product = Product(
        display_name="Async Product",
        description="Created asynchronously",
        product_category="apparel"
    )
    
    created_product = await client.create_product(
        parent=location_path,
        product=product,
        product_id="async_product_1"
    )
    
    print(f"Created product: {created_product.name}")
    
    # List products asynchronously
    response = await client.list_products(parent=location_path)
    for product in response.products:
        print(f"Product: {product.display_name}")

# Run async function
asyncio.run(manage_products_async())