Code für gleichzeitige Anfragen umstrukturieren

// Copyright 2024 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package shopping.merchant.samples.products.v1;

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.core.FixedCredentialsProvider;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.shopping.merchant.products.v1.Availability;
import com.google.shopping.merchant.products.v1.Condition;
import com.google.shopping.merchant.products.v1.InsertProductInputRequest;
import com.google.shopping.merchant.products.v1.ProductAttributes;
import com.google.shopping.merchant.products.v1.ProductInput;
import com.google.shopping.merchant.products.v1.ProductInputsServiceClient;
import com.google.shopping.merchant.products.v1.ProductInputsServiceSettings;
import com.google.shopping.merchant.products.v1.Shipping;
import com.google.shopping.type.Price;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import shopping.merchant.samples.utils.Authenticator;
import shopping.merchant.samples.utils.Config;

/** This class demonstrates how to insert a product input */
public class InsertProductInputAsyncSample {

  private static String getParent(String accountId) {
    return String.format("accounts/%s", accountId);
  }

  private static String generateRandomString() {
    String characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
    Random random = new Random();
    StringBuilder sb = new StringBuilder(8);
    for (int i = 0; i < 8; i++) {
      sb.append(characters.charAt(random.nextInt(characters.length())));
    }
    return sb.toString();
  }

  private static ProductInput createRandomProduct() {
    Price price = Price.newBuilder().setAmountMicros(33_450_000).setCurrencyCode("USD").build();

    Shipping shipping =
        Shipping.newBuilder().setPrice(price).setCountry("GB").setService("1st class post").build();

    Shipping shipping2 =
        Shipping.newBuilder().setPrice(price).setCountry("FR").setService("1st class post").build();

    ProductAttributes attributes =
        ProductAttributes.newBuilder()
            .setTitle("A Tale of Two Cities")
            .setDescription("A classic novel about the French Revolution")
            .setLink("https://exampleWebsite.com/tale-of-two-cities.html")
            .setImageLink("https://exampleWebsite.com/tale-of-two-cities.jpg")
            .setAvailability(Availability.IN_STOCK)
            .setCondition(Condition.NEW)
            .setGoogleProductCategory("Media > Books")
            .addGtins("9780007350896")
            .addShipping(shipping)
            .addShipping(shipping2)
            .build();

    return ProductInput.newBuilder()
        .setContentLanguage("en")
        .setFeedLabel("CH")
        .setOfferId(generateRandomString())
        .setProductAttributes(attributes)
        .build();
  }

  public static void asyncInsertProductInput(Config config, String dataSource) throws Exception {

    // Obtains OAuth token based on the user's configuration.
    GoogleCredentials credential = new Authenticator().authenticate();

    // Creates a channel provider. This provider manages a pool of gRPC channels
    // to enhance throughput for bulk operations. Each individual channel in the pool
    // can handle up to approximately 100 concurrent requests.
    //
    // Channel: A single connection pathway to the service.
    // Pool: A collection of multiple channels managed by this provider.
    //   Requests are distributed across the channels in the pool.
    //
    // We recommend estimating the number of concurrent requests you'll make, divide by 50 (50%
    // utilization of channel capacity), and set the pool size to that number.
    InstantiatingGrpcChannelProvider channelProvider =
        InstantiatingGrpcChannelProvider.newBuilder().setPoolSize(30).build();

    // Creates service settings using the credentials retrieved above.
    ProductInputsServiceSettings productInputsServiceSettings =
        ProductInputsServiceSettings.newBuilder()
            .setCredentialsProvider(FixedCredentialsProvider.create(credential))
            .setTransportChannelProvider(channelProvider)
            .build();

    // Creates parent to identify where to insert the product.
    String parent = getParent(config.getAccountId().toString());

    // Calls the API and catches and prints any network failures/errors.
    try (ProductInputsServiceClient productInputsServiceClient =
        ProductInputsServiceClient.create(productInputsServiceSettings)) {

      // Creates five insert product input requests with random product IDs.
      List<InsertProductInputRequest> requests = new ArrayList<>(5);
      for (int i = 0; i < 5; i++) {
        InsertProductInputRequest request =
            InsertProductInputRequest.newBuilder()
                .setParent(parent)
                // You can only insert products into datasource types of Input "API", and of Type
                // "Primary" or "Supplemental."
                // This field takes the `name` field of the datasource.
                .setDataSource(dataSource)
                // If this product is already owned by another datasource, when re-inserting, the
                // new datasource will take ownership of the product.
                .setProductInput(createRandomProduct())
                .build();

        requests.add(request);
      }

      System.out.println("Sending insert product input requests");
      List<ApiFuture<ProductInput>> futures =
          requests.stream()
              .map(
                  request ->
                      productInputsServiceClient.insertProductInputCallable().futureCall(request))
              .collect(Collectors.toList());

      // Creates callback to handle the responses when all are ready.
      ApiFuture<List<ProductInput>> responses = ApiFutures.allAsList(futures);
      ApiFutures.addCallback(
          responses,
          new ApiFutureCallback<List<ProductInput>>() {
            @Override
            public void onSuccess(List<ProductInput> results) {
              System.out.println("Inserted products below");
              System.out.println(results);
            }

            @Override
            public void onFailure(Throwable throwable) {
              System.out.println(throwable);
            }
          },
          MoreExecutors.directExecutor());

    } catch (Exception e) {
      System.out.println(e);
    }
  }

  public static void main(String[] args) throws Exception {
    Config config = Config.load();
    // Identifies the data source that will own the product input.
    String dataSource = "accounts/" + config.getAccountId() + "/dataSources/{datasourceId}";

    asyncInsertProductInput(config, dataSource);
  }
}
InsertProductInputAsyncSample.java

// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

'use strict';
const fs = require('fs');
const authUtils = require('../../authentication/authenticate.js');
const {
  ProductInputsServiceClient,
} = require('@google-shopping/products').v1;

const {
  protos,
} = require('@google-shopping/products');

const Availability = protos.google.shopping.merchant.products.v1.Availability;
const Condition = protos.google.shopping.merchant.products.v1.Condition;

/**
 * This class demonstrates how to insert a product input asynchronously.
 */

/**
 * Helper function to generate a random string for offerId
 * @returns {string} A sample offerId.
 */
function generateRandomString() {
  const characters =
    'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
  let result = '';
  const length = 8;
  for (let i = 0; i < length; i++) {
    result += characters.charAt(Math.floor(Math.random() * characters.length));
  }
  return result;
}

/**
 * Helper function to create a sample ProductInput object
 * @returns {!object} A sample ProductInput object.
 */
function createRandomProduct() {
  const shippingPrice = {
    amountMicros: 3000000, // 3 USD
    currency_code: 'USD',
  };

  const price = {
    amountMicros: 33450000,  // 33.45 USD
    currency_code: 'USD',
  };

  const shipping = {
    price: shippingPrice,
    country: 'GB',
    service: '1st class post',
  };

  const shipping2 = {
    price: shippingPrice,
    country: 'FR',
    service: '1st class post',
  };

  const attributes = {
    title: 'A Tale of Two Cities',
    description: 'A classic novel about the French Revolution',
    link: 'https://exampleWebsite.com/tale-of-two-cities.html',
    imageLink: 'https://exampleWebsite.com/tale-of-two-cities.jpg',
    availability: Availability.IN_STOCK,
    condition: Condition.NEW,
    googleProductCategory: 'Media > Books',
    gtins: ['9780007350896'],
    shipping: [shipping, shipping2],
    price: price,
  };

  // Construct the ProductInput object
  const productInput = {
    contentLanguage: 'en',
    feedLabel: 'CH',
    offerId: generateRandomString(),
    productAttributes: attributes,
  };

  return productInput;
}

/**
 * Inserts multiple product inputs asynchronously.
 * @param {!object} config - Configuration object.
 * @param {string} dataSource - The data source name.
 */
async function asyncInsertProductInput(config, dataSource) {
  // Read merchant_id from the configuration file.
  const merchantInfo = JSON.parse(
    fs.readFileSync(config.merchantInfoFile, 'utf8')
  );
  const merchantId = merchantInfo.merchantId;

  // Construct the parent resource name string.
  const parent = `accounts/${merchantId}`;

  // Get OAuth2 credentials.
  const authClient = await authUtils.getOrGenerateUserCredentials();

  // Create client options with authentication.
  const options = {authClient: authClient};

  // Creates a pool of clients to enhance throughput for bulk operations.
  // Each individual client in the pool manages its own gRPC channel.
  // We recommend estimating the number of concurrent requests you'll make,
  // divide by 50 (50% utilization of channel capacity), and set the pool size
  // to that number.
  const poolSize = 30;
  const clientPool = [];
  for (let i = 0; i < poolSize; i++) {
    clientPool.push(new ProductInputsServiceClient(options));
  }

  // Create five insert product input requests with random product details.
  const requests = [];
  for (let i = 0; i < 5; i++) {
    const request = {
      parent: parent,
      // You can only insert products into datasource types of Input "API", and
      // of Type "Primary" or "Supplemental."
      // This field takes the `name` field of the datasource, e.g.,
      // accounts/123/dataSources/456
      dataSource: dataSource,
      // If this product is already owned by another datasource, when
      // re-inserting, the new datasource will take ownership of the product.
      productInput: createRandomProduct(),
    };
    requests.push(request);
  }

  console.log('Sending insert product input requests');

  // Create an array of promises by calling the insertProductInput method for
  // each request. Distribute the requests across the client pool to utilize
  // multiple channels.
  const insertPromises = requests.map((request, index) => {
    const client = clientPool[index % poolSize];
    return client.insertProductInput(request);
  });

  // Wait for all insert operations to complete.
  // Promise.all returns an array of results, where each result is the response
  // from the corresponding insertProductInput call (which is the inserted ProductInput).
  // The response from insertProductInput is an array where the first element is the ProductInput.
  const results = await Promise.all(insertPromises);
  const insertedProducts = results.map(result => result[0]); // Extract ProductInput from each response array

  console.log('Inserted products below');
  console.log(insertedProducts);
}

/**
 * Main function to call the async insert product input method.
 */
async function main() {
  // Get configuration settings.
  const config = authUtils.getConfig();
  // Define the data source ID. Replace {datasourceId} with your actual data source ID.
  // The format is accounts/{account_id}/dataSources/{datasource_id}.
  const merchantInfo = JSON.parse(
    fs.readFileSync(config.merchantInfoFile, 'utf8')
  );
  const merchantId = merchantInfo.merchantId;
  const dataSource = `accounts/${merchantId}/dataSources/{datasourceId}`; // Replace {datasourceId}

  try {
    await asyncInsertProductInput(config, dataSource);
  } catch (error) {
    console.log(error);
  }
}

main();
insert_product_input_async_sample.js

<?php
/**
 * Copyright 2025 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

require_once __DIR__ . '/../../../vendor/autoload.php';
require_once __DIR__ . '/../../Authentication/Authentication.php';
require_once __DIR__ . '/../../Authentication/Config.php';
use Google\ApiCore\ApiException;
use Google\Shopping\Merchant\Products\V1\Availability;
use Google\Shopping\Merchant\Products\V1\Condition;
use Google\Shopping\Merchant\Products\V1\ProductAttributes;
use Google\Shopping\Merchant\Products\V1\InsertProductInputRequest;
use Google\Shopping\Merchant\Products\V1\ProductInput;
use Google\Shopping\Merchant\Products\V1\Client\ProductInputsServiceClient;
use Google\Shopping\Merchant\Products\V1\Shipping;
use Google\Shopping\Type\Price;
use React\EventLoop\Loop;
use React\Promise\Promise;
use function React\Promise\all;

/**
 * This class demonstrates how to insert multiple product inputs asynchronously.
 */
class InsertProductInputAsyncSample
{
    /**
     * A helper function to create the parent string for product input operations.
     *
     * @param string $accountId The Merchant Center account ID.
     * @return string The parent resource name format: `accounts/{account_id}`.
     */
    private static function getParent(string $accountId): string
    {
        return sprintf("accounts/%s", $accountId);
    }

    /**
     * Generates a random string of 8 characters.
     *
     * @return string A random alphanumeric string.
     */
    private static function generateRandomString(): string
    {
        $characters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
        $randomString = '';
        $charactersLength = strlen($characters);
        for ($i = 0; $i < 8; $i++) {
            $randomString .= $characters[random_int(0, $charactersLength - 1)];
        }
        return $randomString;
    }

    /**
     * Creates a ProductInput object with randomized offer ID and sample attributes.
     *
     * @return ProductInput A new ProductInput object.
     */
    private static function createRandomProduct(): ProductInput
    {
        // Create a price object for shipping. Amount is in micros.
        // e.g., 33,450,000 micros = $33.45 USD
        $price = new Price([
            'amount_micros' => 33450000,
            'currency_code' => 'USD'
        ]);

        // Create shipping details.
        $shipping = new Shipping([
            'price' => $price,
            'country' => 'GB',
            'service' => '1st class post'
        ]);

        $shipping2 = new Shipping([
            'price' => $price,
            'country' => 'FR',
            'service' => '1st class post'
        ]);

        // Create product attributes.
        $attributes = new ProductAttributes([
            'title' => 'A Tale of Two Cities',
            'description' => 'A classic novel about the French Revolution',
            'link' => 'https://exampleWebsite.com/tale-of-two-cities.html',
            'image_link' => 'https://exampleWebsite.com/tale-of-two-cities.jpg',
            'availability' => Availability::IN_STOCK,
            'condition' => Condition::PBNEW,
            'google_product_category' => 'Media > Books',
            'gtins' => ['9780007350896'],
            'shipping' => [$shipping, $shipping2]
        ]);

        // Create the product input object.
        return new ProductInput([
            'content_language' => 'en',
            'feed_label' => 'LABEL',
            'offer_id' => self::generateRandomString(), // Random offer ID for uniqueness
            'product_attributes' => $attributes
        ]);
    }

    /**
     * Inserts multiple product inputs into the specified account and data source asynchronously.
     *
     * @param array $config Authentication and account configuration.
     * @param string $dataSource The target data source name.
     * Format: `accounts/{account}/dataSources/{datasource}`.
     * @return void
     */
    public static function insertProductInputAsyncSample(array $config, string $dataSource): void
    {
        // Fetches OAuth2 credentials for making API calls.
        $credentials = Authentication::useServiceAccountOrTokenFile();

        // Prepares client options with the fetched credentials.
        $options = ['credentials' => $credentials];

        // Initializes the ProductInputsServiceAsyncClient.
        // This is the key for asynchronous operations.
        $productInputsServiceAsyncClient = new ProductInputsServiceClient($options);

        // Constructs the parent resource string.
        $parent = self::getParent($config['accountId']);

        $promises = [];
        $insertedProductInputs = [];

        print "Sending insert product input requests asynchronously...\n";

        // Create and send 5 insert product input requests asynchronously.
        for ($i = 0; $i < 5; $i++) {
            $productInput = self::createRandomProduct();
            // Create the request object.
            $request = new InsertProductInputRequest([
                'parent' => $parent,
                'data_source' => $dataSource,
                'product_input' => $productInput
            ]);

            // Make the asynchronous API call. This returns a Promise.
            $promise = $productInputsServiceAsyncClient->insertProductInputAsync($request);

            // Attach success and error handlers to the promise.
            $promise->then(
                function (ProductInput $response) use (&$insertedProductInputs) {
                    // This callback is executed when the promise resolves (success).
                    $insertedProductInputs[] = $response;
                    print "Successfully inserted product with offer ID: " . $response->getOfferId() . "\n";
                },
                function (ApiException $e) {
                    // This callback is executed if the promise rejects (failure).
                    echo "ApiException occurred for one of the requests:\n";
                    echo $e;
                }
            );
            $promises[] = $promise;
        }

        // Wait for all promises to settle (either resolve or reject).
        // Reduce::all() creates a single promise that resolves when all input promises resolve.
        // If any promise rejects, the combined promise will reject.
        all($promises)->then(
            function () use (&$insertedProductInputs) {
                print "All asynchronous requests have completed.\n";
                // Print details of all successfully inserted products.
                print "Inserted products below\n";
                foreach ($insertedProductInputs as $p) {
                    print_r($p);
                }
            },
            function ($reason) {
                // This block is executed if any promise in the array rejects.
                echo "One or more asynchronous requests failed.\n";
                if ($reason instanceof ApiException) {
                    echo "API Exception: " . $reason->getMessage() . "\n";
                } else {
                    echo "Error: " . $reason . "\n";
                }
            }
        )->always(function () use ($productInputsServiceAsyncClient) {
            // This 'always' callback ensures the client is closed after all promises settle.
            $productInputsServiceAsyncClient->close();
        });

        // Run the event loop. This is crucial for asynchronous operations to execute.
        // The script will block here until all promises are resolved/rejected or the loop is stopped.
        Loop::run();
    }

    /**
     * Executes the sample code to insert multiple product inputs.
     *
     * @return void
     */
    public function callSample(): void
    {
        $config = Config::generateConfig();

        // Define the data source that will own the product inputs.
        // IMPORTANT: Replace `<DATA_SOURCE_ID>` with your actual data source ID.
        $dataSourceId = '<DATA_SOURCE_ID>';
        $dataSourceName = sprintf(
            "accounts/%s/dataSources/%s",
            $config['accountId'], $dataSourceId
        );

        // Call the method to insert multiple product inputs asynchronously.
        self::insertProductInputAsyncSample($config, $dataSourceName);
    }
}

$sample = new InsertProductInputAsyncSample();
$sample->callSample();
InsertProductInputAsyncSample.php

# -*- coding: utf-8 -*-
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A module to insert product inputs asynchronously."""

import asyncio
import random
import string

from examples.authentication import configuration
from examples.authentication import generate_user_credentials
from google.shopping.merchant_products_v1 import Availability
from google.shopping.merchant_products_v1 import Condition
from google.shopping.merchant_products_v1 import InsertProductInputRequest
from google.shopping.merchant_products_v1 import ProductAttributes
from google.shopping.merchant_products_v1 import ProductInput
from google.shopping.merchant_products_v1 import ProductInputsServiceAsyncClient
from google.shopping.merchant_products_v1 import Shipping
from google.shopping.type import Price

# Read merchant account information from the configuration file.
_ACCOUNT_ID = configuration.Configuration().read_merchant_info()
# The parent account for the product input.
# Format: accounts/{account}
_PARENT = f"accounts/{_ACCOUNT_ID}"


def _generate_random_string(length: int = 8) -> str:
  """Generates a random string of a given length."""
  characters = string.ascii_letters + string.digits
  return "".join(random.choice(characters) for _ in range(length))


def _create_random_product() -> ProductInput:
  """Creates a ProductInput with random elements and predefined attributes."""
  price = Price(amount_micros=33450000, currency_code="USD")

  shipping1 = Shipping(price=price, country="GB", service="1st class post")
  shipping2 = Shipping(price=price, country="FR", service="1st class post")

  attributes = ProductAttributes(
      title="A Tale of Two Cities",
      description="A classic novel about the French Revolution",
      link="https://exampleWebsite.com/tale-of-two-cities.html",
      image_link="https://exampleWebsite.com/tale-of-two-cities.jpg",
      availability=Availability.IN_STOCK,
      condition=Condition.NEW,
      google_product_category="Media > Books",
      gtins=["9780007350896"],
      shipping=[shipping1, shipping2],
  )

  return ProductInput(
      content_language="en",
      feed_label="CH",
      offer_id=_generate_random_string(),
      product_attributes=attributes,
  )


class ClientPool:
  """A simple client pool to distribute requests across multiple clients.

  This implements the Client Pool pattern to enhance throughput for bulk
  operations, mimicking channel pooling.
  """

  def __init__(self, size: int, credentials):
    self._pool = [
        ProductInputsServiceAsyncClient(credentials=credentials)
        for _ in range(size)
    ]
    self._size = size
    self._index = 0

  def get_client(self) -> ProductInputsServiceAsyncClient:
    """Returns the next client in the pool using round-robin."""
    client = self._pool[self._index]
    self._index = (self._index + 1) % self._size
    return client


async def async_insert_product_input(
    client: ProductInputsServiceAsyncClient, request: InsertProductInputRequest
):
  """Inserts product inputs.

  Args:
    client: The ProductInputsServiceAsyncClient to use.
    request: The InsertProductInputRequest to send.

  Returns:
    The response from the insert_product_input request.
  """
  return await client.insert_product_input(request=request)


async def main():
  # The ID of the data source that will own the product input.
  # This is a placeholder and should be replaced with an actual data source ID.
  datasource_id = "<INSERT_DATA_SOURCE_ID_HERE>"
  data_source_name = f"accounts/{_ACCOUNT_ID}/dataSources/{datasource_id}"

  # Gets OAuth Credentials.
  credentials = generate_user_credentials.main()

  # Creates a client pool with 30 clients to handle concurrent requests.
  # We recommend estimating the number of concurrent requests you'll make,
  # divide by 50 (50% utilization of channel capacity), and set the pool size to
  # that number.
  client_pool = ClientPool(size=30, credentials=credentials)

  tasks = []
  for _ in range(5):
    product_input = _create_random_product()
    request = InsertProductInputRequest(
        parent=_PARENT,
        data_source=data_source_name,
        product_input=product_input,
    )

    # Get a client from the pool and create the async task
    client = client_pool.get_client()
    insert_product_task = asyncio.create_task(
        async_insert_product_input(client, request)
    )
    tasks.append(insert_product_task)

  print("Sending insert product input requests")

  try:
    # Await all tasks to complete concurrently and gather their results
    results = await asyncio.gather(*tasks)
    print("Inserted products below")
    print(results)
  except RuntimeError as e:
    # Catch and print any exceptions that occur during the API calls.
    print(e)


if __name__ == "__main__":
  asyncio.run(main())
insert_product_input_async_sample.py