Cloud Function to import product set with bulk import after catalog pushed on GCS

import_product_sets_function

import os
import re

from google.cloud import vision


def call_product_search_vision_api(event, context):
    if bool(re.search('output/product_sets-[\d]+-of-[\d]+\.csv', event['name'])):
        gcs_uri = f'gs://{event["bucket"]}/{event["name"]}'
        _import_product_sets(project_id=os.environ['GCP_PROJECT'], gcs_uri=gcs_uri)


def _import_product_sets(project_id, gcs_uri, location="europe-west1"):
    client = vision.ProductSearchClient()

    location_path = f"projects/{project_id}/locations/{location}"

    gcs_source = vision.ImportProductSetsGcsSource(csv_file_uri=gcs_uri)
    input_config = vision.ImportProductSetsInputConfig(gcs_source=gcs_source)

    response = client.import_product_sets(parent=location_path, input_config=input_config)
    print('Processing operation name: {}'.format(response.operation.name))
    result = response.result()
    print('Processing done.')

    for i, status in enumerate(result.statuses):
        if status.code != 0:
            print('Status code not OK: {}'.format(status.message))