Google Shopping Merchant Quota

from google.shopping.merchant_quota import QuotaServiceClient, QuotaServiceAsyncClient

# v1 (stable) - recommended
from google.shopping.merchant_quota_v1 import QuotaServiceClient, QuotaServiceAsyncClient

# v1beta - for beta features
from google.shopping.merchant_quota_v1beta import QuotaServiceClient, QuotaServiceAsyncClient

from google.shopping.merchant_quota import (
    QuotaGroup,
    MethodDetails,
    ListQuotaGroupsRequest,
    ListQuotaGroupsResponse
)

from google.shopping.merchant_quota import QuotaServiceClient, ListQuotaGroupsRequest

# Create a client with default authentication
client = QuotaServiceClient()

# List quota groups for a merchant account
parent = "accounts/123456789"  # Your merchant account ID
request = ListQuotaGroupsRequest(parent=parent)

# Get paginated results
page_result = client.list_quota_groups(request=request)

# Iterate through quota groups
for quota_group in page_result:
    print(f"Quota Group: {quota_group.name}")
    print(f"Usage: {quota_group.quota_usage}/{quota_group.quota_limit}")
    print(f"Per-minute limit: {quota_group.quota_minute_limit}")
    
    # Show method details
    for method in quota_group.method_details:
        print(f"  Method: {method.method} ({method.version})")
        print(f"  Path: {method.path}")

import asyncio
from google.shopping.merchant_quota import QuotaServiceAsyncClient, ListQuotaGroupsRequest

async def main():
    # Create async client
    async with QuotaServiceAsyncClient() as client:
        parent = "accounts/123456789"
        request = ListQuotaGroupsRequest(parent=parent)
        
        # Get async paginated results
        page_result = await client.list_quota_groups(request=request)
        
        # Iterate through quota groups
        async for quota_group in page_result:
            print(f"Quota Group: {quota_group.name}")
            print(f"Usage: {quota_group.quota_usage}/{quota_group.quota_limit}")

asyncio.run(main())

from typing import Callable, Dict, Optional, Sequence, Tuple, Union
from google.auth import credentials as ga_credentials  
from google.api_core import client_options as client_options_lib
from google.api_core import gapic_v1
from google.api_core import retry as retries
from google.shopping.merchant_quota_v1.services.quota_service import pagers
from google.shopping.merchant_quota_v1.services.quota_service.transports.base import QuotaServiceTransport

try:
    OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault, None]
except AttributeError:
    OptionalRetry = Union[retries.Retry, object, None]

class QuotaServiceClient:
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, QuotaServiceTransport, Callable[..., QuotaServiceTransport]]] = None,
        client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
    ):
        """
        Initialize QuotaService client.
        
        Args:
            credentials: The authorization credentials to attach to requests
            transport: The transport to use for requests
            client_options: Custom options for the client
            client_info: The client info used to send a user-agent string
        """

    def list_quota_groups(
        self,
        request: Optional[Union[ListQuotaGroupsRequest, dict]] = None,
        *,
        parent: Optional[str] = None,
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
        metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
    ) -> pagers.ListQuotaGroupsPager:
        """
        List quota groups for a merchant account.
        
        Args:
            request: The request object containing parent and pagination parameters
            parent: Required. The merchant account (format: accounts/{account})
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional gRPC metadata
            
        Returns:
            Pager for iterating through quota groups
        """

    @classmethod
    def from_service_account_info(
        cls, info: dict, *args, **kwargs
    ) -> "QuotaServiceClient":
        """
        Create client from service account info dictionary.
        
        Args:
            info: Service account info in Google format
            
        Returns:
            QuotaServiceClient instance
        """

    @classmethod
    def from_service_account_file(
        cls, filename: str, *args, **kwargs
    ) -> "QuotaServiceClient":
        """
        Create client from service account JSON file.
        
        Args:
            filename: Path to service account JSON file
            
        Returns:
            QuotaServiceClient instance
        """

    def quota_group_path(self, account: str, group: str) -> str:
        """
        Generate quota group resource path.
        
        Args:
            account: Merchant account ID
            group: Quota group ID
            
        Returns:
            Resource path string (accounts/{account}/quotas/{group})
        """

    @staticmethod
    def parse_quota_group_path(path: str) -> Dict[str, str]:
        """
        Parse quota group resource path.
        
        Args:
            path: Resource path string
            
        Returns:
            Dictionary with 'account' and 'group' keys
        """

    def __enter__(self) -> "QuotaServiceClient":
        """Context manager entry."""

    def __exit__(self, type, value, traceback):
        """Context manager exit."""

from typing import Callable, Optional, Sequence, Tuple, Union
from google.auth import credentials as ga_credentials
from google.api_core import client_options as client_options_lib
from google.api_core import gapic_v1
from google.api_core import retry_async as retries_async
from google.shopping.merchant_quota_v1.services.quota_service import pagers
from google.shopping.merchant_quota_v1.services.quota_service.transports.base import QuotaServiceTransport

try:
    OptionalAsyncRetry = Union[retries_async.AsyncRetry, gapic_v1.method._MethodDefault, None]
except AttributeError:
    OptionalAsyncRetry = Union[retries_async.AsyncRetry, object, None]

class QuotaServiceAsyncClient:
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, QuotaServiceTransport, Callable[..., QuotaServiceTransport]]] = None,
        client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
    ):
        """
        Initialize async QuotaService client.
        
        Args:
            credentials: The authorization credentials to attach to requests
            transport: The transport to use for requests
            client_options: Custom options for the client
            client_info: The client info used to send a user-agent string
        """

    async def list_quota_groups(
        self,
        request: Optional[Union[ListQuotaGroupsRequest, dict]] = None,
        *,
        parent: Optional[str] = None,
        retry: OptionalAsyncRetry = gapic_v1.method.DEFAULT,
        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
        metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
    ) -> pagers.ListQuotaGroupsAsyncPager:
        """
        Asynchronously list quota groups for a merchant account.
        
        Args:
            request: The request object containing parent and pagination parameters
            parent: Required. The merchant account (format: accounts/{account})
            retry: Async retry configuration
            timeout: Request timeout in seconds
            metadata: Additional gRPC metadata
            
        Returns:
            Async pager for iterating through quota groups
        """

    @classmethod
    def from_service_account_info(
        cls, info: dict, *args, **kwargs
    ) -> "QuotaServiceAsyncClient":
        """
        Create async client from service account info dictionary.
        
        Args:
            info: Service account info in Google format
            
        Returns:
            QuotaServiceAsyncClient instance
        """

    @classmethod
    def from_service_account_file(
        cls, filename: str, *args, **kwargs
    ) -> "QuotaServiceAsyncClient":
        """
        Create async client from service account JSON file.
        
        Args:
            filename: Path to service account JSON file
            
        Returns:
            QuotaServiceAsyncClient instance
        """

    async def __aenter__(self) -> "QuotaServiceAsyncClient":
        """Async context manager entry."""

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""

from typing import AsyncIterator, Iterator
from google.shopping.merchant_quota_v1.types.quota import QuotaGroup

class ListQuotaGroupsPager:
    """
    Synchronous iterator for paginated quota group results.
    
    Automatically handles pagination and yields individual QuotaGroup objects.
    """
    def __iter__(self) -> Iterator[QuotaGroup]:
        """Iterate through quota groups."""

class ListQuotaGroupsAsyncPager:
    """
    Asynchronous iterator for paginated quota group results.
    
    Automatically handles pagination and yields individual QuotaGroup objects.
    """
    def __aiter__(self) -> AsyncIterator[QuotaGroup]:
        """Async iterate through quota groups."""

from typing import List

class QuotaGroup:
    """
    Quota group information for methods in the Merchant API.
    
    The quota is shared between all methods in the group. Groups are returned
    even if none of the methods have usage.
    """
    name: str
    """
    Resource name of the quota group.
    Format: accounts/{account}/quotas/{group}
    """
    
    quota_usage: int
    """
    Current quota usage (number of calls made today).
    Daily quotas reset at 12:00 PM midday UTC.
    """
    
    quota_limit: int
    """
    Maximum number of calls allowed per day for the group.
    """
    
    quota_minute_limit: int
    """
    Maximum number of calls allowed per minute for the group.
    """
    
    method_details: List[MethodDetails]
    """
    List of all methods that this quota group applies to.
    """

class MethodDetails:
    """
    Method details for API methods within quota groups.
    """
    method: str
    """
    Method name (e.g., "products.list").
    """
    
    version: str
    """
    API version that the method belongs to.
    """
    
    subapi: str
    """
    Sub-API that the method belongs to.
    """
    
    path: str
    """
    Method path (e.g., "products/v1/productInputs.insert").
    """

class ListQuotaGroupsRequest:
    """
    Request parameters for listing quota groups.
    """
    parent: str
    """
    Required. The merchant account that owns the quota groups.
    Format: accounts/{account}
    """
    
    page_size: int
    """
    Optional. Maximum number of quotas to return (default 500, max 1000).
    Values above 1000 will be coerced to 1000.
    """
    
    page_token: str
    """
    Optional. Token for retrieving subsequent pages.
    All other parameters must match the original call.
    """

class ListQuotaGroupsResponse:
    """
    Response containing quota groups and pagination data.
    """
    quota_groups: List[QuotaGroup]
    """
    Quota groups with usage and limits.
    Groups are sorted in descending order by quota_usage.
    """
    
    next_page_token: str
    """
    Token for retrieving the next page.
    Empty if there are no more pages.
    """
    
    @property
    def raw_page(self):
        """Raw page property for pagination support."""

from google.shopping.merchant_quota import QuotaServiceClient

# Uses Application Default Credentials (ADC)
client = QuotaServiceClient()

client = QuotaServiceClient.from_service_account_file(
    "path/to/service-account.json"
)

service_account_info = {
    "type": "service_account",
    "project_id": "your-project-id",
    # ... other service account fields
}

client = QuotaServiceClient.from_service_account_info(service_account_info)

from google.oauth2 import service_account

credentials = service_account.Credentials.from_service_account_file(
    "path/to/service-account.json"
)

client = QuotaServiceClient(credentials=credentials)

from google.api_core import exceptions
from google.shopping.merchant_quota import QuotaServiceClient

client = QuotaServiceClient()

try:
    parent = "accounts/123456789"
    request = ListQuotaGroupsRequest(parent=parent)
    page_result = client.list_quota_groups(request=request)
    
    for quota_group in page_result:
        print(quota_group.name)
        
except exceptions.NotFound:
    print("Merchant account not found")
except exceptions.PermissionDenied:
    print("Permission denied - check authentication and account access")
except exceptions.InvalidArgument as e:
    print(f"Invalid request parameters: {e}")
except exceptions.GoogleAPIError as e:
    print(f"API error: {e}")

from google.shopping.merchant_quota_v1.services.quota_service import (
    QuotaServiceClient,
    transports
)

# gRPC transport (default)
client = QuotaServiceClient(
    transport=transports.QuotaServiceGrpcTransport()
)

# REST transport
client = QuotaServiceClient(
    transport=transports.QuotaServiceRestTransport()
)

# Async gRPC transport
async_client = QuotaServiceAsyncClient(
    transport=transports.QuotaServiceGrpcAsyncIOTransport()
)

from google.shopping.merchant_quota_v1 import QuotaServiceClient

client = QuotaServiceClient()

from google.shopping.merchant_quota_v1beta import QuotaServiceClient

client = QuotaServiceClient()