Conversion Source Batch Processor

A utility that efficiently processes multiple conversion sources in parallel for a Google Merchant Center account.

Requirements

Build a system that performs batch operations on conversion sources using asynchronous operations for optimal performance.

Core Functionality

The system must:

1. Fetch all conversion sources for a merchant account asynchronously
2. Process multiple conversion sources concurrently (create, retrieve, update operations)
3. Handle pagination when listing conversion sources
4. Properly manage async context and resource cleanup

Specific Behaviors

Batch Creation:

• Accept a list of conversion source configurations (display name, currency code, attribution settings)
• Create multiple Merchant Center Destination conversion sources concurrently
• Return a list of created conversion source identifiers
• It creates 3 conversion sources concurrently and returns their IDs @test

Batch Retrieval:

• Accept a list of conversion source identifiers
• Retrieve details for multiple conversion sources in parallel
• Return a list of conversion source objects with their details
• It retrieves 3 conversion sources in parallel @test

Paginated Listing:

• List all conversion sources for a merchant account using async pagination
• Handle page tokens automatically to iterate through all pages
• Collect and return all conversion sources across pages
• It lists all conversion sources handling pagination @test

@generates

API

import asyncio
from typing import List, Dict, Any

async def batch_create_conversion_sources(
    merchant_id: str,
    conversion_configs: List[Dict[str, Any]]
) -> List[str]:
    """
    Creates multiple conversion sources concurrently.

    Args:
        merchant_id: The merchant account identifier (format: "accounts/{account_id}")
        conversion_configs: List of dicts containing:
            - display_name: str
            - currency_code: str (e.g., "USD")
            - attribution_model: str (e.g., "LAST_CLICK")

    Returns:
        List of created conversion source identifiers (full resource names)
    """
    pass

async def batch_retrieve_conversion_sources(
    conversion_source_ids: List[str]
) -> List[Any]:
    """
    Retrieves multiple conversion sources in parallel.

    Args:
        conversion_source_ids: List of conversion source resource names

    Returns:
        List of conversion source objects
    """
    pass

async def list_all_conversion_sources(merchant_id: str) -> List[Any]:
    """
    Lists all conversion sources for a merchant, handling pagination.

    Args:
        merchant_id: The merchant account identifier (format: "accounts/{account_id}")

    Returns:
        List of all conversion source objects across all pages
    """
    pass

Dependencies { .dependencies }

google-shopping-merchant-conversions { .dependency }

Provides API client for managing conversion sources in Google Merchant Center.

Test Files

• test/batch_create.test.py - Tests concurrent creation of conversion sources
• test/batch_retrieve.test.py - Tests parallel retrieval of conversion sources
• test/paginated_list.test.py - Tests paginated listing with async iteration