tessl/pypi-google-shopping-merchant-reviews
Google Shopping Merchant Reviews API client library for managing merchant and product reviews
—
Pending
merchant-reviews.mddocs/
Merchant Reviews Management
Complete functionality for managing merchant reviews through the Google Shopping Merchant Reviews API. This service enables CRUD operations on merchant reviews, including creation, retrieval, listing with pagination, and deletion.
Capabilities
Client Initialization
Create merchant reviews service clients with authentication and configuration options.
class MerchantReviewsServiceClient:
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, MerchantReviewsServiceTransport, Callable]] = None,
client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = None
):
"""
Initialize the merchant reviews service client.
Args:
credentials: Google authentication credentials
transport: Transport to use for communication ('grpc', 'grpc_asyncio', 'rest', or transport instance)
client_options: Client configuration options
client_info: Client information for user agent
"""
class MerchantReviewsServiceAsyncClient:
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, MerchantReviewsServiceTransport, Callable]] = None,
client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = None
):
"""
Initialize the async merchant reviews service client.
Args:
credentials: Google authentication credentials
transport: Transport to use for communication ('grpc_asyncio' or transport instance)
client_options: Client configuration options
client_info: Client information for user agent
"""Getting Individual Reviews
Retrieve a specific merchant review by its resource name.
def get_merchant_review(
self,
request: Optional[Union[GetMerchantReviewRequest, dict]] = None,
*,
name: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> MerchantReview:
"""
Gets a merchant review.
Args:
request: The request object or dict containing the request fields
name: Required. Resource name format: accounts/{account}/merchantReviews/{merchantReview}
retry: Retry configuration for the request
timeout: Timeout for the request in seconds
metadata: Additional metadata for the request
Returns:
MerchantReview: The requested merchant review
Raises:
google.api_core.exceptions.NotFound: If the review is not found
google.api_core.exceptions.PermissionDenied: If access is denied
"""
async def get_merchant_review(
self,
request: Optional[Union[GetMerchantReviewRequest, dict]] = None,
*,
name: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> MerchantReview:
"""Async version of get_merchant_review."""Listing Reviews with Pagination
List merchant reviews for an account with automatic pagination support.
def list_merchant_reviews(
self,
request: Optional[Union[ListMerchantReviewsRequest, dict]] = None,
*,
parent: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ListMerchantReviewsPager:
"""
Lists merchant reviews for an account.
Args:
request: The request object or dict containing the request fields
parent: Required. Parent resource format: accounts/{account}
retry: Retry configuration for the request
timeout: Timeout for the request in seconds
metadata: Additional metadata for the request
Returns:
ListMerchantReviewsPager: Pager for iterating through reviews
Raises:
google.api_core.exceptions.InvalidArgument: If parent format is invalid
google.api_core.exceptions.PermissionDenied: If access is denied
"""
async def list_merchant_reviews(
self,
request: Optional[Union[ListMerchantReviewsRequest, dict]] = None,
*,
parent: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> ListMerchantReviewsAsyncPager:
"""Async version of list_merchant_reviews."""Inserting Reviews
Insert or replace a merchant review.
def insert_merchant_review(
self,
request: Optional[Union[InsertMerchantReviewRequest, dict]] = None,
*,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> MerchantReview:
"""
Inserts or replaces a merchant review.
Args:
request: The request object containing parent, merchant_review, and data_source
retry: Retry configuration for the request
timeout: Timeout for the request in seconds
metadata: Additional metadata for the request
Returns:
MerchantReview: The inserted merchant review
Raises:
google.api_core.exceptions.InvalidArgument: If request data is invalid
google.api_core.exceptions.PermissionDenied: If access is denied
"""
async def insert_merchant_review(
self,
request: Optional[Union[InsertMerchantReviewRequest, dict]] = None,
*,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> MerchantReview:
"""Async version of insert_merchant_review."""Deleting Reviews
Delete a merchant review permanently.
def delete_merchant_review(
self,
request: Optional[Union[DeleteMerchantReviewRequest, dict]] = None,
*,
name: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> None:
"""
Deletes a merchant review.
Args:
request: The request object or dict containing the request fields
name: Required. Resource name format: accounts/{account}/merchantReviews/{merchantReview}
retry: Retry configuration for the request
timeout: Timeout for the request in seconds
metadata: Additional metadata for the request
Raises:
google.api_core.exceptions.NotFound: If the review is not found
google.api_core.exceptions.PermissionDenied: If access is denied
"""
async def delete_merchant_review(
self,
request: Optional[Union[DeleteMerchantReviewRequest, dict]] = None,
*,
name: Optional[str] = None,
retry = gapic_v1.method.DEFAULT,
timeout = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> None:
"""Async version of delete_merchant_review."""Helper Methods
Resource path construction and parsing utilities.
@classmethod
def merchant_review_path(cls, account: str, name: str) -> str:
"""
Returns a fully-qualified merchant review string.
Args:
account: Account ID
name: Merchant review name/ID
Returns:
str: Resource path in format accounts/{account}/merchantReviews/{merchantReview}
"""
@classmethod
def parse_merchant_review_path(cls, path: str) -> Dict[str, str]:
"""
Parses a merchant review path.
Args:
path: Merchant review resource path
Returns:
Dict[str, str]: Dictionary with 'account' and 'merchant_review' keys
Raises:
ValueError: If path format is invalid
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> MerchantReviewsServiceClient:
"""
Creates a client from a service account file.
Args:
filename: Path to service account JSON file
*args: Additional arguments to pass to constructor
**kwargs: Additional keyword arguments to pass to constructor
Returns:
MerchantReviewsServiceClient: Configured client instance
"""
@classmethod
def from_service_account_info(
cls,
info: dict,
*args,
**kwargs
) -> MerchantReviewsServiceClient:
"""
Creates a client from service account info.
Args:
info: Service account info dictionary
*args: Additional arguments to pass to constructor
**kwargs: Additional keyword arguments to pass to constructor
Returns:
MerchantReviewsServiceClient: Configured client instance
"""Usage Examples
Basic Review Management
from google.shopping.merchant_reviews import (
MerchantReviewsServiceClient,
MerchantReview,
MerchantReviewAttributes,
InsertMerchantReviewRequest
)
from google.auth import default
from google.protobuf.timestamp_pb2 import Timestamp
import time
# Initialize client
credentials, project = default()
client = MerchantReviewsServiceClient(credentials=credentials)
# Create a new merchant review
review_attributes = MerchantReviewAttributes(
merchant_id="your-merchant-id",
merchant_display_name="Your Business Name",
content="Great service and fast delivery!",
rating=5.0,
min_rating=1,
max_rating=5,
reviewer_username="satisfied_customer",
reviewer_id="reviewer123",
collection_method=MerchantReviewAttributes.CollectionMethod.AFTER_FULFILLMENT
)
# Set review time
review_time = Timestamp()
review_time.FromSeconds(int(time.time()))
review_attributes.review_time = review_time
merchant_review = MerchantReview(
merchant_review_id="review-12345",
merchant_review_attributes=review_attributes
)
# Insert the review
request = InsertMerchantReviewRequest(
parent="accounts/your-account-id",
merchant_review=merchant_review,
data_source="accounts/your-account-id/dataSources/your-datasource-id"
)
result = client.insert_merchant_review(request=request)
print(f"Inserted review: {result.name}")
# List reviews
reviews = client.list_merchant_reviews(parent="accounts/your-account-id")
for review in reviews:
print(f"Review ID: {review.merchant_review_id}")
print(f"Rating: {review.merchant_review_attributes.rating}")
# Get specific review
review_name = f"accounts/your-account-id/merchantReviews/review-12345"
specific_review = client.get_merchant_review(name=review_name)
print(f"Retrieved review rating: {specific_review.merchant_review_attributes.rating}")
# Delete review
client.delete_merchant_review(name=review_name)
print("Review deleted")Async Operations
import asyncio
from google.shopping.merchant_reviews import MerchantReviewsServiceAsyncClient
async def manage_reviews_async():
# Initialize async client
client = MerchantReviewsServiceAsyncClient()
# List reviews asynchronously
reviews = await client.list_merchant_reviews(
parent="accounts/your-account-id"
)
# Iterate through reviews
async for review in reviews:
print(f"Review ID: {review.merchant_review_id}")
if review.merchant_review_attributes.rating < 3.0:
# Handle low-rated reviews
print(f"Low rating detected: {review.merchant_review_attributes.rating}")
# Run async function
asyncio.run(manage_reviews_async())Pagination Handling
# Handle pagination manually
parent = "accounts/your-account-id"
request = ListMerchantReviewsRequest(
parent=parent,
page_size=100 # Process 100 reviews at a time
)
while True:
response = client.list_merchant_reviews(request=request)
# Process current page
for review in response.merchant_reviews:
print(f"Processing review: {review.merchant_review_id}")
# Check if there are more pages
if not response.next_page_token:
break
# Set up request for next page
request.page_token = response.next_page_token
# Or use the pager (recommended)
pager = client.list_merchant_reviews(parent=parent)
for review in pager:
print(f"Review ID: {review.merchant_review_id}")Types
# Request types
class GetMerchantReviewRequest:
name: str # Required: accounts/{account}/merchantReviews/{merchantReview}
class DeleteMerchantReviewRequest:
name: str # Required: accounts/{account}/merchantReviews/{merchantReview}
class ListMerchantReviewsRequest:
parent: str # Required: accounts/{account}
page_size: int # Optional: max 1000
page_token: str # Optional: pagination token
class InsertMerchantReviewRequest:
parent: str # Required: accounts/{account}
merchant_review: MerchantReview # Required: review to insert
data_source: str # Required: accounts/{account}/dataSources/{datasource}
# Response types
class ListMerchantReviewsResponse:
merchant_reviews: MutableSequence[MerchantReview]
next_page_token: str
@property
def raw_page(self) -> ListMerchantReviewsResponse: ...Install with Tessl CLI
npx tessl i tessl/pypi-google-shopping-merchant-reviews